diff --git a/litellm/a2a_protocol/main.py b/litellm/a2a_protocol/main.py index 3a1d2c70b12..0ee2906da9a 100644 --- a/litellm/a2a_protocol/main.py +++ b/litellm/a2a_protocol/main.py @@ -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. diff --git a/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py b/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py index 34c18b4f0bd..636051d34ce 100644 --- a/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py @@ -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() diff --git a/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py b/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py index 34f5fdc7c5f..86ce8cb5c09 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py +++ b/litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py @@ -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( diff --git a/litellm/proxy/management_endpoints/key_management_endpoints.py b/litellm/proxy/management_endpoints/key_management_endpoints.py index be70136e4fe..a3f65b7315b 100644 --- a/litellm/proxy/management_endpoints/key_management_endpoints.py +++ b/litellm/proxy/management_endpoints/key_management_endpoints.py @@ -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. diff --git a/litellm/proxy/management_endpoints/model_management_endpoints.py b/litellm/proxy/management_endpoints/model_management_endpoints.py index 4e46a070ec2..673ea829723 100644 --- a/litellm/proxy/management_endpoints/model_management_endpoints.py +++ b/litellm/proxy/management_endpoints/model_management_endpoints.py @@ -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 diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index d6358c638e7..1663907b556 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -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")