mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-15 23:31:29 +00:00
Fix code qa
This commit is contained in:
Sameer Kankute
parent
6c1118e438
commit
9b76f10746
3 changed files with 256 additions and 156 deletions
|
|
@ -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())]
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -1870,6 +1870,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",
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue