scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-26 06:03:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 35

fix(responses): comply with strict ruff and type discipline gates

Browse source
This commit is contained in:
ashishtransfirm 2026-10-18 20:20:31 +05:30
parent 75d724b76f
commit cf7a19f9f3
1 changed files with 32 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

52
litellm/responses/litellm_completion_transformation/streaming_iterator.py
View file

@ -942,24 +942,37 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
}
}
"""
item_dict: Final[dict[str, object]] = {
"id": reasoning_item_id,
"type": "reasoning",
"status": "completed",
"summary": [
{
"type": "summary_text",
"text": reasoning_content,
}
],
}
if encrypted_content:
item_dict["encrypted_content"] = encrypted_content
reasoning_item: Final = (
BaseLiteLLMOpenAIResponseObject(
id=reasoning_item_id,
type="reasoning",
status="completed",
summary=[
{
"type": "summary_text",
"text": reasoning_content,
}
],
encrypted_content=encrypted_content,
)
if encrypted_content
else BaseLiteLLMOpenAIResponseObject(
id=reasoning_item_id,
type="reasoning",
status="completed",
summary=[
{
"type": "summary_text",
"text": reasoning_content,
}
],
)
)
return OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=self._reasoning_index(),
sequence_number=sequence_number,
item=BaseLiteLLMOpenAIResponseObject(**item_dict),
item=reasoning_item,
)
def return_default_done_events(
@ -1091,20 +1104,19 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def _extract_streamed_thinking_blocks_encrypted(self) -> str | None:
if self._cached_encrypted_reasoning_content is not None:
return self._cached_encrypted_reasoning_content
chunks = [c for c in self.collected_chat_completion_chunks if c.get("choices")]
chunks: Final = [c for c in self.collected_chat_completion_chunks if c.get("choices")]
if not any((c["choices"][0].get("delta") or {}).get("thinking_blocks") for c in chunks):
return None
for c in chunks:
if not c.get("model"):
c["model"] = self.model or "unknown"
model_name: Final = self.model or "unknown"
prepared_chunks: Final = [c if c.get("model") else {**c, "model": model_name} for c in chunks]
try:
built = stream_chunk_builder(chunks=chunks)
built: Final = stream_chunk_builder(chunks=prepared_chunks)
if built and built.choices and hasattr(built.choices[0], "message"):
self._cached_encrypted_reasoning_content = LiteLLMCompletionResponsesConfig._encode_thinking_blocks(
built.choices[0].message
)
return self._cached_encrypted_reasoning_content
except Exception:
except (KeyError, ValueError, TypeError, AttributeError, IndexError):
return None
return None