From c1d907f355fd0a98259bc622e71101ea750e09b7 Mon Sep 17 00:00:00 2001 From: Sameer Kankute Date: Fri, 24 Apr 2026 17:13:25 +0530 Subject: [PATCH] Fix code qa --- .../transformation.py | 194 ++++++++++------- .../llms/chatgpt/responses/transformation.py | 201 +++++++++++------- provider_endpoints_support.json | 17 ++ 3 files changed, 256 insertions(+), 156 deletions(-) diff --git a/litellm/completion_extras/litellm_responses_transformation/transformation.py b/litellm/completion_extras/litellm_responses_transformation/transformation.py index 6f2819ce902..18a274b8b0c 100644 --- a/litellm/completion_extras/litellm_responses_transformation/transformation.py +++ b/litellm/completion_extras/litellm_responses_transformation/transformation.py @@ -584,6 +584,109 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): return choices + @classmethod + def _parse_raw_sse_chunk(cls, chunk: str) -> Optional[Dict[str, Any]]: + stripped_chunk = ( + CustomStreamWrapper._strip_sse_data_from_chunk(chunk.strip()) or "" + ).strip() + if ( + not stripped_chunk + or stripped_chunk == "[DONE]" + or stripped_chunk.startswith("event:") + ): + return None + try: + parsed_chunk = json.loads(stripped_chunk) + except json.JSONDecodeError: + return None + if not isinstance(parsed_chunk, dict): + return None + return parsed_chunk + + @classmethod + def _extract_output_from_completed_event( + cls, parsed_chunk: Dict[str, Any] + ) -> Optional[List[Dict[str, Any]]]: + response_payload = parsed_chunk.get("response") + if not isinstance(response_payload, dict): + return None + response_output = response_payload.get("output") + if not isinstance(response_output, list) or len(response_output) == 0: + return None + return cast(List[Dict[str, Any]], response_output) + + @classmethod + def _update_recovered_output_items( + cls, parsed_chunk: Dict[str, Any], recovered_output_items: Dict[int, Dict[str, Any]] + ) -> None: + item = parsed_chunk.get("item") + if not isinstance(item, dict): + return + try: + output_index = int(parsed_chunk.get("output_index")) + except (TypeError, ValueError): + output_index = len(recovered_output_items) + recovered_output_items[output_index] = item + + @classmethod + def _update_recovered_text_only_items( + cls, + parsed_chunk: Dict[str, Any], + recovered_output_items: Dict[int, Dict[str, Any]], + recovered_text_only_items: Dict[int, Dict[str, Any]], + ) -> None: + text = parsed_chunk.get("text") + if not isinstance(text, str): + return + + try: + output_index = int(parsed_chunk.get("output_index")) + except (TypeError, ValueError): + output_index = len(recovered_text_only_items) + + item = recovered_output_items.get(output_index) or recovered_text_only_items.get( + output_index + ) + if item is None: + item = { + "type": "message", + "id": parsed_chunk.get("item_id") or f"msg_{output_index}", + "role": "assistant", + "status": "completed", + "content": [], + } + recovered_text_only_items[output_index] = item + + content = item.setdefault("content", []) + if not isinstance(content, list): + return + + try: + content_index = int(parsed_chunk.get("content_index")) + except (TypeError, ValueError): + content_index = len(content) + + while len(content) <= content_index: + content.append( + { + "type": "output_text", + "text": "", + "annotations": [], + } + ) + + content_item = content[content_index] + if not isinstance(content_item, dict): + content_item = {} + content[content_index] = content_item + + content_item["type"] = "output_text" + content_item["text"] = text + if parsed_chunk.get("annotations") is not None: + content_item["annotations"] = parsed_chunk["annotations"] + else: + content_item.setdefault("annotations", []) + @classmethod def _recover_output_items_from_raw_sse( cls, raw_sse: Optional[str] @@ -595,97 +698,28 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge): recovered_text_only_items: Dict[int, Dict[str, Any]] = {} for chunk in raw_sse.splitlines(): - stripped_chunk = ( - CustomStreamWrapper._strip_sse_data_from_chunk(chunk.strip()) or "" - ).strip() - if ( - not stripped_chunk - or stripped_chunk == "[DONE]" - or stripped_chunk.startswith("event:") - ): - continue - - try: - parsed_chunk = json.loads(stripped_chunk) - except json.JSONDecodeError: - continue - - if not isinstance(parsed_chunk, dict): + parsed_chunk = cls._parse_raw_sse_chunk(chunk) + if parsed_chunk is None: continue event_type = parsed_chunk.get("type") if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: - response_payload = parsed_chunk.get("response") - if isinstance(response_payload, dict): - response_output = response_payload.get("output") - if isinstance(response_output, list) and len(response_output) > 0: - return cast(List[Dict[str, Any]], response_output) + recovered_output = cls._extract_output_from_completed_event(parsed_chunk) + if recovered_output is not None: + return recovered_output continue if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: - item = parsed_chunk.get("item") - if not isinstance(item, dict): - continue - try: - output_index = int(parsed_chunk.get("output_index")) - except (TypeError, ValueError): - output_index = len(recovered_output_items) - recovered_output_items[output_index] = item + cls._update_recovered_output_items(parsed_chunk, recovered_output_items) continue if event_type == ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE: - text = parsed_chunk.get("text") - if not isinstance(text, str): - continue - - try: - output_index = int(parsed_chunk.get("output_index")) - except (TypeError, ValueError): - output_index = len(recovered_text_only_items) - - item = recovered_output_items.get( - output_index - ) or recovered_text_only_items.get(output_index) - if item is None: - item = { - "type": "message", - "id": parsed_chunk.get("item_id") or f"msg_{output_index}", - "role": "assistant", - "status": "completed", - "content": [], - } - recovered_text_only_items[output_index] = item - - content = item.setdefault("content", []) - if not isinstance(content, list): - continue - - try: - content_index = int(parsed_chunk.get("content_index")) - except (TypeError, ValueError): - content_index = len(content) - - while len(content) <= content_index: - content.append( - { - "type": "output_text", - "text": "", - "annotations": [], - } - ) - - content_item = content[content_index] - if not isinstance(content_item, dict): - content_item = {} - content[content_index] = content_item - - content_item["type"] = "output_text" - content_item["text"] = text - if parsed_chunk.get("annotations") is not None: - content_item["annotations"] = parsed_chunk["annotations"] - else: - content_item.setdefault("annotations", []) + cls._update_recovered_text_only_items( + parsed_chunk=parsed_chunk, + recovered_output_items=recovered_output_items, + recovered_text_only_items=recovered_text_only_items, + ) if recovered_output_items: return [item for _, item in sorted(recovered_output_items.items())] diff --git a/litellm/llms/chatgpt/responses/transformation.py b/litellm/llms/chatgpt/responses/transformation.py index 4874f08d05e..9c2a55c3bfe 100644 --- a/litellm/llms/chatgpt/responses/transformation.py +++ b/litellm/llms/chatgpt/responses/transformation.py @@ -111,101 +111,150 @@ class ChatGPTResponsesAPIConfig(OpenAIResponsesAPIConfig): raw_response: Any, logging_obj: Any, ): - content_type = (raw_response.headers or {}).get("content-type", "") body_text = raw_response.text or "" - if "text/event-stream" not in content_type.lower(): - trimmed_body = body_text.lstrip() - if not ( - trimmed_body.startswith("event:") - or trimmed_body.startswith("data:") - or "\nevent:" in body_text - or "\ndata:" in body_text - ): - return super().transform_response_api_response( - model=model, - raw_response=raw_response, - logging_obj=logging_obj, - ) + if not self._should_parse_as_sse(raw_response=raw_response, body_text=body_text): + return super().transform_response_api_response( + model=model, + raw_response=raw_response, + logging_obj=logging_obj, + ) logging_obj.post_call( original_response=raw_response.text, additional_args={"complete_input_dict": {}}, ) - completed_response = None - error_message = None - streamed_output_items: Dict[int, dict] = {} - for chunk in body_text.splitlines(): - stripped_chunk = CustomStreamWrapper._strip_sse_data_from_chunk(chunk) - if not stripped_chunk: - continue - stripped_chunk = stripped_chunk.strip() - if not stripped_chunk: - continue - if stripped_chunk == STREAM_SSE_DONE_STRING: - break - try: - parsed_chunk = json.loads(stripped_chunk) - except json.JSONDecodeError: - continue - if not isinstance(parsed_chunk, dict): - continue - event_type = parsed_chunk.get("type") - if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: - item = parsed_chunk.get("item") - output_index = parsed_chunk.get("output_index") - if isinstance(item, dict): - try: - index = int(output_index) - except (TypeError, ValueError): - index = len(streamed_output_items) - streamed_output_items[index] = item - continue - if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: - response_payload = parsed_chunk.get("response") - if isinstance(response_payload, dict): - response_payload = dict(response_payload) - if not response_payload.get("output") and streamed_output_items: - response_payload["output"] = [ - item for _, item in sorted(streamed_output_items.items()) - ] - if "created_at" in response_payload: - response_payload["created_at"] = _safe_convert_created_field( - response_payload["created_at"] - ) - try: - completed_response = ResponsesAPIResponse(**response_payload) - except Exception: - completed_response = ResponsesAPIResponse.model_construct( - **response_payload - ) - break - if event_type in ( - ResponsesAPIStreamEvents.RESPONSE_FAILED, - ResponsesAPIStreamEvents.ERROR, - ): - error_obj = parsed_chunk.get("error") or ( - parsed_chunk.get("response") or {} - ).get("error") - if error_obj is not None: - if isinstance(error_obj, dict): - error_message = error_obj.get("message") or str(error_obj) - else: - error_message = str(error_obj) - + completed_response, error_message = self._extract_completed_response_from_sse( + body_text=body_text + ) if completed_response is None: raise OpenAIError( message=error_message or raw_response.text, status_code=raw_response.status_code, ) + self._attach_response_headers( + completed_response=completed_response, raw_response=raw_response + ) + return completed_response + + def _should_parse_as_sse(self, raw_response: Any, body_text: str) -> bool: + content_type = (raw_response.headers or {}).get("content-type", "") + if "text/event-stream" in content_type.lower(): + return True + trimmed_body = body_text.lstrip() + return bool( + trimmed_body.startswith("event:") + or trimmed_body.startswith("data:") + or "\nevent:" in body_text + or "\ndata:" in body_text + ) + + def _extract_completed_response_from_sse( + self, body_text: str + ) -> tuple[Optional[ResponsesAPIResponse], Optional[str]]: + completed_response = None + error_message = None + streamed_output_items: Dict[int, dict] = {} + for chunk in body_text.splitlines(): + parsed_chunk = self._parse_sse_json_chunk(chunk) + if parsed_chunk is None: + continue + if parsed_chunk == STREAM_SSE_DONE_STRING: + break + + event_type = parsed_chunk.get("type") + if event_type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE: + self._record_output_item_chunk( + parsed_chunk=parsed_chunk, streamed_output_items=streamed_output_items + ) + continue + + if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED: + completed_response = self._build_completed_response_from_chunk( + parsed_chunk=parsed_chunk, streamed_output_items=streamed_output_items + ) + break + + if event_type in ( + ResponsesAPIStreamEvents.RESPONSE_FAILED, + ResponsesAPIStreamEvents.ERROR, + ): + error_message = self._extract_error_message(parsed_chunk) + + return completed_response, error_message + + def _parse_sse_json_chunk(self, chunk: str) -> Optional[Any]: + stripped_chunk = CustomStreamWrapper._strip_sse_data_from_chunk(chunk) + if not stripped_chunk: + return None + stripped_chunk = stripped_chunk.strip() + if not stripped_chunk: + return None + if stripped_chunk == STREAM_SSE_DONE_STRING: + return STREAM_SSE_DONE_STRING + try: + parsed_chunk = json.loads(stripped_chunk) + except json.JSONDecodeError: + return None + if not isinstance(parsed_chunk, dict): + return None + return parsed_chunk + + def _record_output_item_chunk( + self, parsed_chunk: Dict[str, Any], streamed_output_items: Dict[int, dict] + ) -> None: + item = parsed_chunk.get("item") + output_index = parsed_chunk.get("output_index") + if not isinstance(item, dict): + return + try: + index = int(output_index) + except (TypeError, ValueError): + index = len(streamed_output_items) + streamed_output_items[index] = item + + def _build_completed_response_from_chunk( + self, parsed_chunk: Dict[str, Any], streamed_output_items: Dict[int, dict] + ) -> Optional[ResponsesAPIResponse]: + response_payload = parsed_chunk.get("response") + if not isinstance(response_payload, dict): + return None + response_payload = dict(response_payload) + if not response_payload.get("output") and streamed_output_items: + response_payload["output"] = [ + item for _, item in sorted(streamed_output_items.items()) + ] + if "created_at" in response_payload: + response_payload["created_at"] = _safe_convert_created_field( + response_payload["created_at"] + ) + try: + return ResponsesAPIResponse(**response_payload) + except Exception: + return ResponsesAPIResponse.model_construct(**response_payload) + + def _extract_error_message(self, parsed_chunk: Dict[str, Any]) -> Optional[str]: + error_obj = parsed_chunk.get("error") or (parsed_chunk.get("response") or {}).get( + "error" + ) + if error_obj is None: + return None + if isinstance(error_obj, dict): + return error_obj.get("message") or str(error_obj) + return str(error_obj) + + def _attach_response_headers( + self, + completed_response: ResponsesAPIResponse, + raw_response: Any, + ) -> None: raw_headers = dict(raw_response.headers) processed_headers = process_response_headers(raw_headers) if not hasattr(completed_response, "_hidden_params"): setattr(completed_response, "_hidden_params", {}) completed_response._hidden_params["additional_headers"] = processed_headers completed_response._hidden_params["headers"] = raw_headers - return completed_response def get_complete_url( self, diff --git a/provider_endpoints_support.json b/provider_endpoints_support.json index 6f23c87f911..463c705c7aa 100644 --- a/provider_endpoints_support.json +++ b/provider_endpoints_support.json @@ -1853,6 +1853,23 @@ "rerank": false } }, + "reducto": { + "display_name": "Reducto (`reducto`)", + "url": "https://docs.litellm.ai/docs/providers/reducto", + "endpoints": { + "chat_completions": false, + "messages": false, + "responses": false, + "embeddings": false, + "image_generations": false, + "audio_transcriptions": false, + "audio_speech": false, + "moderations": false, + "batches": false, + "rerank": false, + "ocr": true + } + }, "replicate": { "display_name": "Replicate (`replicate`)", "url": "https://docs.litellm.ai/docs/providers/replicate",