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

fix(anthropic): type the block and role reads so the basedpyright budget stays flat on main

Browse source
This commit is contained in:
Tan Nguyen 2026-10-09 02:31:27 +07:00
parent cabcc738b4
commit 036eec9e06
1 changed files with 16 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -230,7 +230,7 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES
def _index_of_block_accepting_cache_control(content: list[object], on_a_tool_message: bool) -> int | None:
def _index_of_block_accepting_cache_control(content: Sequence[object], on_a_tool_message: bool) -> int | None:
"""Last block a marker written here reaches the provider on, searched from the end.
An empty text block is replaced by a placeholder unless it sits on a tool message,
@ -240,14 +240,22 @@ def _index_of_block_accepting_cache_control(content: list[object], on_a_tool_mes
block = content[index]
if not isinstance(block, dict):
continue
if block.get("type") in ANTHROPIC_BLOCK_TYPES_WITHOUT_CACHE_CONTROL:
fields = cast(dict[str, object], block) # cast-ok: a runtime dict whose value types are not known here
if fields.get("type") in ANTHROPIC_BLOCK_TYPES_WITHOUT_CACHE_CONTROL:
continue
if block.get("type") == "text" and not block.get("text") and not on_a_tool_message:
if fields.get("type") == "text" and not fields.get("text") and not on_a_tool_message:
continue
return index
return None
def _role_of(message: object) -> object:
"""The ``role`` of a message dict, or None for anything else."""
if not isinstance(message, dict):
return None
return cast(dict[str, object], message).get("role") # cast-ok: a runtime dict whose value types are not known here
def _message_accepts_cache_control(message: object) -> bool:
"""Whether a marker written on this message reaches the provider."""
if not isinstance(message, dict):
@ -258,7 +266,8 @@ def _message_accepts_cache_control(message: object) -> bool:
if isinstance(content, str):
return content != "" or on_a_tool_message
if isinstance(content, list):
return _index_of_block_accepting_cache_control(content, on_a_tool_message) is not None
blocks: Final = cast(Sequence[object], content) # cast-ok: a runtime list whose block types are not known here
return _index_of_block_accepting_cache_control(blocks, on_a_tool_message) is not None
return False
@ -580,7 +589,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
candidates: Final = tuple(
index
for index, message in enumerate(messages)
if targetted_role is None or message.get("role") == targetted_role
if targetted_role is None or _role_of(message) == targetted_role
)
free: Final = frozenset(
index
@ -650,8 +659,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
message["cache_control"] = control
# 2. list of objects - the last block that accepts a marker, per Anthropic spec
elif isinstance(message_content, list):
blocks: Final = cast(Sequence[object], message_content) # cast-ok: TypedDict blocks read as objects
target_index: Final = _index_of_block_accepting_cache_control(
message_content, on_a_tool_message=message.get("role") == "tool"
blocks, on_a_tool_message=_role_of(message) == "tool"
)
if target_index is not None:
message_content[target_index]["cache_control"] = control # pyright: ignore[reportGeneralTypeIssues] # loose runtime dict