scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-14 04:09:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 97

refactor(types): replace Any with proven types in 6 files (#46042)

Browse source 
* refactor(types): replace Any with proven types in 8 files

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* refactor(types): keep azure search and google_genai seams untyped

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-10-13 10:00:52 +00:00 • committed by GitHub
parent a5d05355b6
commit 77005080f3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 59 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/a2a_protocol/main.py
View file

@ -634,7 +634,7 @@ async def asend_message_streaming(
proxy_server_request: dict[str, object] | None = None,
agent_extra_headers: dict[str, str] | None = None,
**kwargs: object,
) -> AsyncIterator[Any]:
) -> AsyncIterator["SendStreamingMessageResponse | Mapping[str, object]"]:
"""
Async: Send a streaming message to an A2A agent.

63
litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
View file

@ -214,36 +214,24 @@ class _CombinedChunkSplitter:
pieces: Final = tuple(copy.deepcopy(chunk) for _ in present_groups)
for index, (piece, group) in enumerate(zip(pieces, present_groups)):
copied_delta = piece.choices[0].delta
fields = {field: value for field in group if (value := getattr(copied_delta, field, None))}
fields = _CombinedChunkSplitter._normalize_reasoning_fields(fields)
reasoning_content = (
getattr(copied_delta, "reasoning_content", None) if "reasoning_content" in group else None
)
thinking_blocks = getattr(copied_delta, "thinking_blocks", None) if "thinking_blocks" in group else None
unsigned_text = _CombinedChunkSplitter._unsigned_thinking_text(thinking_blocks) if thinking_blocks else None
if unsigned_text:
reasoning_content = unsigned_text
thinking_blocks = None
role = getattr(copied_delta, "role", None) if index == 0 else None
piece.choices[0].delta = Delta(role=role, **fields)
piece.choices[0].delta = Delta(
role=role,
reasoning_content=reasoning_content if reasoning_content else None,
thinking_blocks=thinking_blocks if thinking_blocks else None,
content=(copied_delta.content if "content" in group and copied_delta.content else None),
tool_calls=(copied_delta.tool_calls if "tool_calls" in group and copied_delta.tool_calls else None),
)
return pieces
@staticmethod
def _normalize_reasoning_fields(fields: "dict[str, Any]") -> "dict[str, Any]":
"""Collapse signature-less ``thinking_blocks`` into ``reasoning_content``.
The block opener seeds a ``thinking_blocks`` start body with the full
thinking text while the delta re-emits it, so SSE accumulators would
collect it twice; the ``reasoning_content`` branch opens an empty body.
Signature-carrying blocks are kept intact so ``signature_delta``
suppression of the full-text snapshot still applies.
"""
thinking_blocks: Final = fields.get("thinking_blocks")
if not thinking_blocks:
return fields
if any(block.get("signature") for block in thinking_blocks if isinstance(block, dict)):
return fields
thinking_text: Final = "".join(
block.get("thinking") or ""
for block in thinking_blocks
if isinstance(block, dict) and block.get("type") == "thinking"
)
if not thinking_text:
return fields
return {"reasoning_content": thinking_text}
@staticmethod
def _split(chunk: "ModelResponseStream") -> "list[ModelResponseStream]":
"""Return ``[chunk]``, or ``[content_chunk, finish_chunk]`` if combined."""
@ -270,6 +258,27 @@ class _CombinedChunkSplitter:
def __iter__(self) -> "Iterator[ModelResponseStream]":
return self
@staticmethod
def _unsigned_thinking_text(
blocks: Sequence[Mapping[str, Any]],
) -> str | None:
"""Return the combined thinking text when no block carries a signature.
The block opener seeds a ``thinking_blocks`` start body with the full
thinking text while the delta re-emits it, so SSE accumulators would
collect it twice; the ``reasoning_content`` branch opens an empty body.
Signature-carrying blocks are kept intact so ``signature_delta``
suppression of the full-text snapshot still applies.
"""
if any(block.get("signature") for block in blocks if isinstance(block, dict)):
return None
thinking_text: Final = "".join(
block.get("thinking") or ""
for block in blocks
if isinstance(block, dict) and block.get("type") == "thinking"
)
return thinking_text or None
def __next__(self) -> "ModelResponseStream":
if self._buffer:
return self._buffer.popleft()

2
litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py
View file

@ -1229,7 +1229,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
or attempt >= _BEDROCK_APPLY_GUARDRAIL_MAX_THROTTLE_RETRIES
):
raise
await asyncio.sleep(_BEDROCK_APPLY_GUARDRAIL_BASE_BACKOFF_SECONDS * (2**attempt))
await asyncio.sleep(_BEDROCK_APPLY_GUARDRAIL_BASE_BACKOFF_SECONDS * (1 << attempt))
raise HTTPException(status_code=500, detail="Bedrock guardrail throttle retries exhausted")
async def _post_apply_guardrail_content(

6
litellm/proxy/management_endpoints/key_management_endpoints.py
View file

@ -21,7 +21,7 @@ import traceback
from collections.abc import Awaitable, Callable, Iterator, Mapping, Sequence
from datetime import datetime, timedelta, timezone
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, TypeVar, cast
from typing import TYPE_CHECKING, Final, Literal, Optional, Protocol, TypeVar, cast
import fastapi
import yaml
@ -6478,7 +6478,7 @@ async def reset_key_spend_fn(
None,
description="The litellm-changed-by header enables tracking of actions performed by authorized users on behalf of other users, providing an audit trail for accountability",
),
) -> dict[str, Any]:
) -> dict[str, object]:
try:
from litellm.proxy.proxy_server import (
hash_token,
@ -7022,7 +7022,7 @@ async def key_aliases(
size: int = Query(50, ge=1, le=100, description="Page size"),
search: str | None = Query(None, description="Search key aliases (case-insensitive partial match)"),
team_id: str | None = Query(None, description="Filter aliases to keys belonging to this team"),
) -> dict[str, Any]:
) -> dict[str, object]:
"""
Lists key aliases with pagination and optional search.

8
litellm/proxy/management_endpoints/model_management_endpoints.py
View file

@ -467,8 +467,12 @@ def _effective_complexity_router_config(
).effective
def _dump_without_none(model: BaseModel) -> Mapping[str, object]:
return model.model_dump(exclude_none=True)
def _decrypted_litellm_params(litellm_params: GenericLiteLLMParams) -> Mapping[str, object]:
dumped: Final[Mapping[str, object]] = litellm_params.model_dump(exclude_none=True)
dumped: Final = _dump_without_none(litellm_params)
return MappingProxyType(
{
name: (
@ -1037,7 +1041,7 @@ def update_db_model(
_raise_if_ptu_cost_attribution_disabled(updated_patch.model_info.model_dump(exclude_none=True))
merged_model_name: Final = updated_patch.model_name or db_model.model_name
merged_litellm_params: Final = db_model.litellm_params.model_dump(exclude_none=True)
stored_model_info: Final = db_model.model_info.model_dump(exclude_none=True)
stored_model_info: Final = _dump_without_none(db_model.model_info)
echoed_pricing: Final = echoed_cost_map_pricing_fields(stored_model_info)
merged_model_info: Final[dict[str, object]] = {
k: v for k, v in stored_model_info.items() if k not in echoed_pricing

17
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -2322,7 +2322,7 @@ def _rewrite_vertex_live_setup_model(text_data: str, setup_model_rewriter: Calla
if setup_model_rewriter is None:
return text_data
try:
message: Final = json.loads(text_data)
message: Final[object] = json.loads(text_data)
except json.JSONDecodeError:
return text_data
if not isinstance(message, dict):
@ -2357,7 +2357,7 @@ def _resolved_vertex_live_setup(
def _json_object_frame(frame: str | bytes) -> dict[str, object] | None:
try:
decoded: Final = json.loads(frame if isinstance(frame, str) else frame.decode("utf-8"))
decoded: Final[object] = json.loads(frame if isinstance(frame, str) else frame.decode("utf-8"))
except (json.JSONDecodeError, UnicodeDecodeError):
return None
return decoded if isinstance(decoded, dict) else None
@ -2844,9 +2844,16 @@ async def websocket_passthrough_request(
await websocket.close()
def _content_type_header(response: httpx.Response) -> str:
try:
return response.headers["content-type"]
except KeyError:
return ""
def _is_streaming_response(response: httpx.Response) -> bool:
_content_type: Final = response.headers.get("content-type")
if _content_type is not None and "text/event-stream" in _content_type:
_content_type: Final[str] = _content_type_header(response)
if "text/event-stream" in _content_type:
return True
return False
@ -2894,7 +2901,7 @@ def _should_buffer_passthrough_response(response: httpx.Response) -> bool:
"""
if response.status_code >= 400:
return True
content_type_header: Final[str] = response.headers.get("content-type", "")
content_type_header: Final[str] = _content_type_header(response)
media_type: Final = content_type_header.split(";")[0].strip().lower()
return (
media_type in ("", "application/json")