Merge pull request #9419 from BerriAI/litellm_streaming_o1_pro

[Feat] OpenAI o1-pro Responses API streaming support

litellm/llms/base_llm/responses/transformation.py

@@ -7,7 +7,6 @@ import httpx
 from litellm.types.llms.openai import (
     ResponseInputParam,
     ResponsesAPIOptionalRequestParams,
-    ResponsesAPIRequestParams,
     ResponsesAPIResponse,
     ResponsesAPIStreamingResponse,
 )
@@ -97,7 +96,7 @@ class BaseResponsesAPIConfig(ABC):
         response_api_optional_request_params: Dict,
         litellm_params: GenericLiteLLMParams,
         headers: dict,
-    ) -> ResponsesAPIRequestParams:
+    ) -> Dict:
         pass
 
     @abstractmethod
@@ -131,3 +130,12 @@ class BaseResponsesAPIConfig(ABC):
             message=error_message,
             headers=headers,
         )
+
+    def should_fake_stream(
+        self,
+        model: Optional[str],
+        stream: Optional[bool],
+        custom_llm_provider: Optional[str] = None,
+    ) -> bool:
+        """Returns True if litellm should fake a stream for the given model and stream value"""
+        return False

litellm/llms/custom_httpx/llm_http_handler.py

@@ -20,6 +20,7 @@ from litellm.llms.custom_httpx.http_handler import (
 )
 from litellm.responses.streaming_iterator import (
     BaseResponsesAPIStreamingIterator,
+    MockResponsesAPIStreamingIterator,
     ResponsesAPIStreamingIterator,
     SyncResponsesAPIStreamingIterator,
 )
@@ -978,6 +979,7 @@ class BaseLLMHTTPHandler:
         timeout: Optional[Union[float, httpx.Timeout]] = None,
         client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
         _is_async: bool = False,
+        fake_stream: bool = False,
     ) -> Union[
         ResponsesAPIResponse,
         BaseResponsesAPIStreamingIterator,
@@ -1003,6 +1005,7 @@ class BaseLLMHTTPHandler:
                 extra_body=extra_body,
                 timeout=timeout,
                 client=client if isinstance(client, AsyncHTTPHandler) else None,
+                fake_stream=fake_stream,
             )
 
         if client is None or not isinstance(client, HTTPHandler):
@@ -1051,14 +1054,27 @@ class BaseLLMHTTPHandler:
         try:
             if stream:
                 # For streaming, use stream=True in the request
+                if fake_stream is True:
+                    stream, data = self._prepare_fake_stream_request(
+                        stream=stream,
+                        data=data,
+                        fake_stream=fake_stream,
+                    )
                 response = sync_httpx_client.post(
                     url=api_base,
                     headers=headers,
                     data=json.dumps(data),
                     timeout=timeout
                     or response_api_optional_request_params.get("timeout"),
-                    stream=True,
+                    stream=stream,
                 )
+                if fake_stream is True:
+                    return MockResponsesAPIStreamingIterator(
+                        response=response,
+                        model=model,
+                        logging_obj=logging_obj,
+                        responses_api_provider_config=responses_api_provider_config,
+                    )
 
                 return SyncResponsesAPIStreamingIterator(
                     response=response,
@@ -1100,6 +1116,7 @@ class BaseLLMHTTPHandler:
         extra_body: Optional[Dict[str, Any]] = None,
         timeout: Optional[Union[float, httpx.Timeout]] = None,
         client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
+        fake_stream: bool = False,
     ) -> Union[ResponsesAPIResponse, BaseResponsesAPIStreamingIterator]:
         """
         Async version of the responses API handler.
@@ -1145,22 +1162,36 @@ class BaseLLMHTTPHandler:
                 "headers": headers,
             },
         )
-
         # Check if streaming is requested
         stream = response_api_optional_request_params.get("stream", False)
 
         try:
             if stream:
                 # For streaming, we need to use stream=True in the request
+                if fake_stream is True:
+                    stream, data = self._prepare_fake_stream_request(
+                        stream=stream,
+                        data=data,
+                        fake_stream=fake_stream,
+                    )
+
                 response = await async_httpx_client.post(
                     url=api_base,
                     headers=headers,
                     data=json.dumps(data),
                     timeout=timeout
                     or response_api_optional_request_params.get("timeout"),
-                    stream=True,
+                    stream=stream,
                 )
 
+                if fake_stream is True:
+                    return MockResponsesAPIStreamingIterator(
+                        response=response,
+                        model=model,
+                        logging_obj=logging_obj,
+                        responses_api_provider_config=responses_api_provider_config,
+                    )
+
                 # Return the streaming iterator
                 return ResponsesAPIStreamingIterator(
                     response=response,
@@ -1177,6 +1208,7 @@ class BaseLLMHTTPHandler:
                     timeout=timeout
                     or response_api_optional_request_params.get("timeout"),
                 )
+
         except Exception as e:
             raise self._handle_error(
                 e=e,
@@ -1189,6 +1221,21 @@ class BaseLLMHTTPHandler:
             logging_obj=logging_obj,
         )
 
+    def _prepare_fake_stream_request(
+        self,
+        stream: bool,
+        data: dict,
+        fake_stream: bool,
+    ) -> Tuple[bool, dict]:
+        """
+        Handles preparing a request when `fake_stream` is True.
+        """
+        if fake_stream is True:
+            stream = False
+            data.pop("stream", None)
+            return stream, data
+        return stream, data
+
     def _handle_error(
         self,
         e: Exception,

litellm/llms/openai/responses/transformation.py

@@ -65,10 +65,12 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
         response_api_optional_request_params: Dict,
         litellm_params: GenericLiteLLMParams,
         headers: dict,
-    ) -> ResponsesAPIRequestParams:
+    ) -> Dict:
         """No transform applied since inputs are in OpenAI spec already"""
-        return ResponsesAPIRequestParams(
-            model=model, input=input, **response_api_optional_request_params
+        return dict(
+            ResponsesAPIRequestParams(
+                model=model, input=input, **response_api_optional_request_params
+            )
         )
 
     def transform_response_api_response(
@@ -188,3 +190,27 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
             raise ValueError(f"Unknown event type: {event_type}")
 
         return model_class
+
+    def should_fake_stream(
+        self,
+        model: Optional[str],
+        stream: Optional[bool],
+        custom_llm_provider: Optional[str] = None,
+    ) -> bool:
+        if stream is not True:
+            return False
+        if model is not None:
+            try:
+                if (
+                    litellm.utils.supports_native_streaming(
+                        model=model,
+                        custom_llm_provider=custom_llm_provider,
+                    )
+                    is False
+                ):
+                    return True
+            except Exception as e:
+                verbose_logger.debug(
+                    f"Error getting model info in OpenAIResponsesAPIConfig: {e}"
+                )
+        return False