diff --git a/litellm/integrations/anthropic_cache_control_hook.py b/litellm/integrations/anthropic_cache_control_hook.py index 4302b24ca33..4e473c2f3f0 100644 --- a/litellm/integrations/anthropic_cache_control_hook.py +++ b/litellm/integrations/anthropic_cache_control_hook.py @@ -9,6 +9,7 @@ Supported for both `v1/chat/completions` (via the prompt-management hook) and """ +import bisect import copy import itertools import os @@ -21,12 +22,16 @@ from urllib.parse import urlparse from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError from litellm._logging import verbose_logger +from litellm.exceptions import UnsupportedParamsError from litellm.integrations.custom_logger import CustomLogger from litellm.integrations.custom_prompt_management import CustomPromptManagement from litellm.integrations.prompt_management_base import PromptManagementClient from litellm.litellm_core_utils.prompt_templates.common_utils import ( + PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT, cache_control_as_prompt_cache_breakpoint, + prompt_cache_breakpoint_as_cache_control, with_prompt_cache_breakpoint, + without_prompt_cache_breakpoints, ) from litellm.litellm_core_utils.prompt_templates.factory import ( tool_call_ids_answered_by_client, @@ -327,6 +332,12 @@ def _part_with_cache_control_renamed(part: object) -> object: return cache_control_as_prompt_cache_breakpoint(_CONTENT_PART_ADAPTER.validate_python(part)) +def _part_with_prompt_cache_breakpoint_renamed(part: object, drop_params: bool) -> object: + if not _carries_prompt_cache_breakpoint(part): + return part + return prompt_cache_breakpoint_as_cache_control(_CONTENT_PART_ADAPTER.validate_python(part), drop_params) + + def configured_injection_points(value: object) -> Sequence[CacheControlInjectionPoint]: if not isinstance(value, (list, tuple)): return () @@ -551,6 +562,10 @@ def _carries_cache_breakpoint(block: object) -> bool: return any(_attribute_or_key(block, key) is not None for key in CACHE_BREAKPOINT_KEYS) +def _carries_cache_control(block: object) -> bool: + return _attribute_or_key(block, "cache_control") is not None + + def _attribute_or_key(value: object, key: str) -> object | None: if hasattr(value, key): return _ATTRIBUTE_VALUE.validate_python(getattr(value, key)) @@ -630,6 +645,12 @@ def _builds_anthropic_messages(model: str, custom_llm_provider: str | None) -> b return isinstance(chat_config, AnthropicConfig) and not isinstance(chat_config, OpenAIGPTConfig) +def _translates_prompt_cache_breakpoint(model: str, custom_llm_provider: str | None) -> bool: + return (custom_llm_provider or _served_provider_for_model(model)) == "bedrock" or _builds_anthropic_messages( + model, custom_llm_provider + ) + + def _custom_llm_provider(non_default_params: Mapping[str, object], point: Mapping[str, object]) -> str | None: provider: Final = non_default_params.get("custom_llm_provider") or point.get(CUSTOM_LLM_PROVIDER_STAMP) return provider if isinstance(provider, str) else None @@ -673,10 +694,79 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool: return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES +def _carries_prompt_cache_breakpoint(block: object) -> bool: + return _accepts_prompt_cache_breakpoint(block) and _attribute_or_key(block, "prompt_cache_breakpoint") is not None + + def _may_carry_prompt_cache_breakpoint(message: Mapping[str, object]) -> bool: return message.get("role") != "assistant" +def _tool_result_carries_prompt_cache_breakpoint(message: Mapping[str, object]) -> bool: + return message.get("role") == "tool" and any( + _carries_prompt_cache_breakpoint(part) for part in _as_object_list(message.get("content")) or () + ) + + +def _part_without_prompt_cache_breakpoint(part: object) -> object: + if not _carries_prompt_cache_breakpoint(part): + return part + return { + key: value + for key, value in _CONTENT_PART_ADAPTER.validate_python(part).items() + if key != "prompt_cache_breakpoint" + } + + +def _tool_result_without_prompt_cache_breakpoints(message: AllMessageValues) -> AllMessageValues: + if not _tool_result_carries_prompt_cache_breakpoint(message): + return message + parts: Final = _as_object_list(_attribute_or_key(message, "content")) or [] + return cast( # cast-ok: the same tool message with its part marks removed + AllMessageValues, {**message, "content": [_part_without_prompt_cache_breakpoint(part) for part in parts]} + ) + + +def _created_cache_control(original: object, translated: object) -> bool: + return ( + _attribute_or_key(original, "cache_control") is None + and _attribute_or_key(translated, "cache_control") is not None + ) + + +def _created_cache_controls(original: object, translated: object) -> tuple[bool, ...]: + if translated is original: + return () + original_parts: Final = _as_object_list(_attribute_or_key(original, "content")) or [] + translated_parts: Final = _as_object_list(_attribute_or_key(translated, "content")) or [] + return tuple(_created_cache_control(before, after) for before, after in zip(original_parts, translated_parts)) + + +def _part_without_cache_control(part: object) -> object: + return {key: value for key, value in _CONTENT_PART_ADAPTER.validate_python(part).items() if key != "cache_control"} + + +def _message_dropping_created_cache_controls_between( + message: AllMessageValues, created: Sequence[bool], marks_before: int, first_dropped: int, first_kept_after: int +) -> AllMessageValues: + if not any(created): + return message + parts: Final = _as_object_list(_attribute_or_key(message, "content")) or [] + ordinals: Final = itertools.accumulate((int(flag) for flag in created), initial=marks_before) + return cast( # cast-ok: the same message with its middle translated marks removed + AllMessageValues, + { + **message, + "content": [ + _part_without_cache_control(part) + if is_created and first_dropped <= ordinal < first_kept_after + else part + for part, is_created, ordinal in zip(parts, created, ordinals) + ], + }, + ) + + # Set by a caller whose message list is not the one that goes upstream -- today the # Responses API layer, whose `instructions` only becomes a system message further down. # Tells this hook to hand role-targeted points to the pass holding the final messages @@ -762,9 +852,6 @@ class AnthropicCacheControlHook(CustomPromptManagement): if not injection_points: return model, messages, non_default_params - # Create a deep copy of messages to avoid modifying the original list - processed_messages = copy.deepcopy(messages) - message_points: Final = tuple( cast(CacheControlMessageInjectionPoint, point) for point in injection_points @@ -775,6 +862,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): request_params: Final = cast(Mapping[str, object], non_default_params) # cast-ok: base hook's bare dict custom_llm_provider: Final = _custom_llm_provider(request_params, injection_points[0]) stamped_dialect: Final = injection_points[0].get("_litellm_openai_dialect") + model_info: Final = routed_model_info(request_params.get("model_info")) openai_dialect: Final = ( stamped_dialect if isinstance(stamped_dialect, bool) @@ -783,9 +871,14 @@ class AnthropicCacheControlHook(CustomPromptManagement): custom_llm_provider, request_params.get("api_base") or request_params.get("base_url"), request_params.get("prompt_cache_options"), - routed_model_info(request_params.get("model_info")), + model_info, ) ) + processed_messages = copy.deepcopy( + [without_prompt_cache_breakpoints(message) for message in messages] + if not openai_dialect and message_points and _translates_prompt_cache_breakpoint(model, custom_llm_provider) + else messages + ) # A provisional message list defers every role-targeted point to the pass holding # the final one: a role with no message here may have one there, and settling all # of them in one pass is what lets config order decide the shared breakpoint @@ -876,7 +969,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): ) ) if hosted_flag is not None: - return hosted_flag + return hosted_flag and not _translates_prompt_cache_breakpoint(model, custom_llm_provider) if not supports_openai_prompt_cache_breakpoint(model, model_info=model_info): return False if (custom_llm_provider or AnthropicCacheControlHook._resolve_provider(model)) != "openai": @@ -929,6 +1022,19 @@ class AnthropicCacheControlHook(CustomPromptManagement): ) return int(wire_cache_control is not None) + tool_blocks + envelope_blocks + @staticmethod + def tools_keeping_cache_control_on_chat(tools: object) -> tuple[object, ...]: + return tuple( + tool for tool in _as_object_list(tools) or () if not _chat_transform_drops_tool_cache_control(tool) + ) + + @staticmethod + def count_chat_external_cache_breakpoints(optional_params: object) -> int: + params: Final = _validated_object_mapping(optional_params) or {} + return AnthropicCacheControlHook.count_external_cache_breakpoints( + _as_object_list(params.get("tools")), params.get("cache_control") + ) + @staticmethod def count_external_cache_breakpoints_on_messages_route( tools: Iterable[object] | None, cache_control: object, request_kwargs: object @@ -1084,6 +1190,16 @@ class AnthropicCacheControlHook(CustomPromptManagement): ) return message_count + content_count + tool_call_count + @staticmethod + def count_converse_cache_points(message: AllMessageValues) -> int: + content: Final = _as_object_list(_attribute_or_key(message, "content")) + tool_calls: Final = _as_object_list(_attribute_or_key(message, "tool_calls")) or [] + marked_parts: Final = sum(1 for block in content or () if _carries_cache_control(block)) + marked_tool_calls: Final = sum(1 for tool_call in tool_calls if _carries_cache_control(tool_call)) + if message.get("role") == "tool": + return int(_carries_cache_control(message) or marked_parts > 0) + return (int(_carries_cache_control(message)) if content is None else marked_parts) + marked_tool_calls + @staticmethod def _message_has_cache_control(message: AllMessageValues, answered_tool_ids: Container[str]) -> bool: """Return True if the message already carries any cache_control.""" @@ -1142,6 +1258,92 @@ class AnthropicCacheControlHook(CustomPromptManagement): return renamed return AnthropicCacheControlHook._insert_prompt_cache_breakpoint_in_message(renamed) + @staticmethod + def prompt_cache_breakpoints_as_cache_control(message: AllMessageValues, drop_params: bool) -> AllMessageValues: + """Every OpenAI ``prompt_cache_breakpoint`` on a part OpenAI accepts one on moved to Anthropic's + ``cache_control``. Marks with no OpenAI slot are left for the caller to strip.""" + if not _may_carry_prompt_cache_breakpoint(message): + return message + parts: Final = _as_object_list(_attribute_or_key(message, "content")) + if parts is None or not any(_carries_prompt_cache_breakpoint(part) for part in parts): + return message + return cast( # cast-ok: the same message with its part marks renamed + AllMessageValues, + {**message, "content": [_part_with_prompt_cache_breakpoint_renamed(part, drop_params) for part in parts]}, + ) + + @staticmethod + def prompt_cache_breakpoints_dropped_from_tool_results( + messages: Sequence[AllMessageValues], drop_params: bool, model: str + ) -> tuple[AllMessageValues, ...]: + if not any(_tool_result_carries_prompt_cache_breakpoint(message) for message in messages): + return tuple(messages) + if not drop_params: + raise UnsupportedParamsError( + message=( + f"{model} takes no cache breakpoint on a tool result, so prompt_cache_breakpoint on a tool " + f"message has no equivalent. Remove it, or {PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT} to drop it." + ), + status_code=400, + ) + verbose_logger.debug("Dropping prompt_cache_breakpoint from tool results for %s under drop_params", model) + return tuple(_tool_result_without_prompt_cache_breakpoints(message) for message in messages) + + @staticmethod + def prompt_cache_breakpoints_as_capped_cache_control( + messages: Sequence[AllMessageValues], + drop_params: bool, + sent_breakpoints: Callable[[AllMessageValues], int], + external_breakpoints: Callable[[], int], + ) -> tuple[AllMessageValues, ...]: + """``prompt_cache_breakpoints_as_cache_control`` over the whole request, held to the provider's + breakpoint cap, counting each message by the breakpoints ``sent_breakpoints`` says its provider sends. + Under ``drop_params`` the first and last translated marks fill the slots the client's own + ``cache_control`` marks leave, the first half taking the odd slot, so the stable prefix and the newest + turns stay cached, and without it a request over the cap is rejected.""" + translated: Final = tuple( + AnthropicCacheControlHook.prompt_cache_breakpoints_as_cache_control(message, drop_params) + for message in messages + ) + if all(after is before for before, after in zip(messages, translated)): + return translated + created: Final = tuple(_created_cache_controls(before, after) for before, after in zip(messages, translated)) + created_marks: Final = sum(sum(flags) for flags in created) + if created_marks == 0: + return translated + external: Final = external_breakpoints() + breakpoints: Final = sum(map(sent_breakpoints, translated)) + external + excess: Final = breakpoints - MAX_CACHE_CONTROL_BLOCKS + if excess <= 0: + return translated + if not drop_params: + raise UnsupportedParamsError( + message=( + f"This model takes at most {MAX_CACHE_CONTROL_BLOCKS} cache breakpoints per request, and this " + f"request has {breakpoints} once its prompt_cache_breakpoint marks are translated. Remove {excess} " + f"of them, or {PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT} to keep the first and last of them." + ), + status_code=400, + ) + verbose_logger.debug("Dropping the middle prompt_cache_breakpoint marks past the cap under drop_params") + marks_before: Final = tuple(itertools.accumulate((sum(flags) for flags in created), initial=0)) + + def capped_request(kept_marks: int) -> tuple[AllMessageValues, ...]: + first_dropped: Final = (kept_marks + 1) // 2 + first_kept_after: Final = created_marks - kept_marks // 2 + return tuple( + _message_dropping_created_cache_controls_between( + message, flags, before, first_dropped, first_kept_after + ) + for message, flags, before in zip(translated, created, marks_before) + ) + + def over_cap(kept_marks: int) -> bool: + return sum(map(sent_breakpoints, capped_request(kept_marks))) + external > MAX_CACHE_CONTROL_BLOCKS + + first_over_cap: Final = bisect.bisect_left(range(max(0, created_marks - excess) + 1), True, key=over_cap) + return capped_request(max(0, first_over_cap - 1)) + @staticmethod def _insert_prompt_cache_breakpoint_in_message(message: AllMessageValues) -> AllMessageValues: if not _may_carry_prompt_cache_breakpoint(message): @@ -1487,9 +1689,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): configured: Final = configured_injection_points(non_default_params.get("cache_control_injection_points")) if configured: - tools_keeping_marks: Final = tuple( - tool for tool in tools or () if not _chat_transform_drops_tool_cache_control(tool) - ) + tools_keeping_marks: Final = AnthropicCacheControlHook.tools_keeping_cache_control_on_chat(tools) stamped: Final = AnthropicCacheControlHook._stamped_for_prompt_hook( configured, AnthropicCacheControlHook.count_external_cache_breakpoints( @@ -1782,9 +1982,17 @@ def messages_within_cache_control_cap( Tool hints and a top-level ``cache_control`` stay on the wire, so they are reserved first. """ - reserved_marks: Final = AnthropicCacheControlHook.count_external_cache_breakpoints( - _validated_object_list(tools), cache_control, request_kwargs + return messages_within_reserved_cache_control_cap( + messages, + AnthropicCacheControlHook.count_external_cache_breakpoints( + _validated_object_list(tools), cache_control, request_kwargs + ), ) + + +def messages_within_reserved_cache_control_cap( + messages: Sequence[_MessageT], reserved_marks: int +) -> Sequence[_MessageT]: kept: Final = newest_marks_within_cache_control_limit(messages, reserved_marks) dropped: Final = sum(map(cache_control_count, messages)) - sum(map(cache_control_count, kept)) if dropped > 0: diff --git a/litellm/litellm_core_utils/prompt_templates/common_utils.py b/litellm/litellm_core_utils/prompt_templates/common_utils.py index 6287dc17e8f..6ecab166dcd 100644 --- a/litellm/litellm_core_utils/prompt_templates/common_utils.py +++ b/litellm/litellm_core_utils/prompt_templates/common_utils.py @@ -21,12 +21,14 @@ from pydantic import TypeAdapter, ValidationError import litellm from litellm import verbose_logger +from litellm.exceptions import UnsupportedParamsError from litellm.litellm_core_utils.batch_utils import InMemoryFile from litellm.litellm_core_utils.core_helpers import as_str_mapping from litellm.types.llms.openai import ( AllMessageValues, ChatCompletionAssistantMessage, ChatCompletionAudioObject, + ChatCompletionCachedContent, ChatCompletionFileObject, ChatCompletionImageObject, ChatCompletionPromptCacheBreakpoint, @@ -2183,6 +2185,9 @@ def check_is_function_call(logging_obj: "LoggingClass") -> bool: _MarkedT: Final = TypeVar("_MarkedT", bound=Mapping[str, object]) +PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT: Final = ( + "set `drop_params: true` on the request or in litellm_settings (`litellm.drop_params = True` in the SDK)" +) def with_prompt_cache_breakpoint(target: _MarkedT, marker: object) -> _MarkedT: @@ -2262,6 +2267,51 @@ def cache_control_as_prompt_cache_breakpoint(block: _MarkedT) -> _MarkedT: ) +def prompt_cache_breakpoint_as_cache_control(block: _MarkedT, drop_params: bool) -> _MarkedT: + """OpenAI's ``prompt_cache_breakpoint`` mark rewritten as Anthropic's ``cache_control`` on the same block. + A ``cache_control`` the client already wrote on the block wins, TTL included.""" + marker: Final = block.get("prompt_cache_breakpoint") + if marker is None: + return block + unmarked: Final = {key: value for key, value in block.items() if key != "prompt_cache_breakpoint"} + control: Final = ( + None + if block.get("cache_control") is not None + else _cache_control_for_prompt_cache_breakpoint(marker, drop_params) + ) + marked: Final = unmarked if control is None else {**unmarked, "cache_control": control} + return cast(_MarkedT, marked) # cast-ok: same block shape as the input with the marker key renamed + + +def _cache_control_for_prompt_cache_breakpoint(marker: object, drop_params: bool) -> ChatCompletionCachedContent | None: + try: + breakpoint_mark: Final = _PROMPT_CACHE_BREAKPOINT.validate_python(marker) + except ValidationError: + if not drop_params: + raise UnsupportedParamsError( + message=( + f"prompt_cache_breakpoint must be {{'mode': 'explicit'}} with an optional ttl of '30m' or '12h', got " + f"{marker!r}. Fix the marker, or " + f"{PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT} to drop it." + ), + status_code=400, + ) + verbose_logger.debug("Dropping malformed prompt_cache_breakpoint %r under drop_params", marker) + return None + ttl: Final = breakpoint_mark.get("ttl") + if ttl is not None and not drop_params: + raise UnsupportedParamsError( + message=( + f"This model's prompt cache lasts 5m, or 1h where it offers one, so prompt_cache_breakpoint " + f"ttl={ttl!r} has no equivalent. Remove the ttl, {PROMPT_CACHE_BREAKPOINT_DROP_PARAMS_HINT} to send " + "a 5m breakpoint instead, or send cache_control {'type': 'ephemeral', 'ttl': '1h'} on the block " + "where the model offers a 1h cache." + ), + status_code=400, + ) + return ChatCompletionCachedContent(type="ephemeral") + + _BREAKPOINT_BLOCK_LIST_KEYS: Final = frozenset({"content", "tool_calls"}) _BREAKPOINT_FIELDS: Final = TypeAdapter(Mapping[str, object]) _BREAKPOINT_BLOCKS: Final = TypeAdapter(list[object]) diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index 1977e3e2a7d..6b96df8f879 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -2370,6 +2370,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): messages, optional_params.get("tools"), optional_params.get("cache_control"), optional_params ) + def translates_prompt_cache_breakpoint(self, model: str) -> bool: + return True + def transform_request( self, model: str, @@ -2383,19 +2386,32 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): """ ## VALIDATE REQUEST """Anthropic requires ``tools`` when messages include tool blocks; LiteLLM injects a dummy tool if omitted (no ``modify_params`` needed).""" + from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook, cache_control_count from litellm.litellm_core_utils.prompt_templates.factory import ( anthropic_messages_pt, ) drop_params: Final = drops_unsupported_params(litellm_params.get("drop_params")) + translated_messages: Final = ( + list( + AnthropicCacheControlHook.prompt_cache_breakpoints_as_capped_cache_control( + messages, + drop_params, + cache_control_count, + lambda: AnthropicCacheControlHook.count_chat_external_cache_breakpoints(optional_params), + ) + ) + if self.translates_prompt_cache_breakpoint(model) + else messages + ) capped_messages: Final = ( list( self._messages_within_cache_control_cap( - messages, _REQUEST_PARAMS_ADAPTER.validate_python(optional_params) + translated_messages, _REQUEST_PARAMS_ADAPTER.validate_python(optional_params) ) ) if drop_params - else messages + else translated_messages ) if "tools" not in optional_params and has_tool_call_blocks(capped_messages): diff --git a/litellm/llms/bedrock/chat/converse_transformation.py b/litellm/llms/bedrock/chat/converse_transformation.py index f2985237554..056f34f1bf4 100644 --- a/litellm/llms/bedrock/chat/converse_transformation.py +++ b/litellm/llms/bedrock/chat/converse_transformation.py @@ -23,6 +23,11 @@ from litellm.constants import ( BEDROCK_MIN_THINKING_BUDGET_TOKENS, RESPONSE_FORMAT_TOOL_NAME, ) +from litellm.integrations.anthropic_cache_control_hook import ( + MAX_CACHE_CONTROL_BLOCKS, + AnthropicCacheControlHook, + configured_injection_points, +) from litellm.litellm_core_utils.core_helpers import ( as_str_mapping, drops_unsupported_params, @@ -131,6 +136,7 @@ from ..common_utils import ( bedrock_capability_model, bedrock_converse_supports_parallel_tool_use_config, bedrock_model_accepts_cache_points, + bedrock_model_known_to_accept_cache_points, bedrock_reasoning_effort_disabled, bedrock_rejects_stop_sequences, get_anthropic_beta_from_headers, @@ -204,6 +210,11 @@ def _has_native_text_format(output_config: object) -> bool: return False +def _cache_mark_kept(block: Mapping[str, object], keeps_cache_marks: bool) -> Mapping[str, object]: + cache_control: Final = block.get("cache_control") + return {"cache_control": cache_control} if keeps_cache_marks and cache_control is not None else {} + + class AmazonConverseConfig(BaseConfig): """ Reference - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html @@ -243,7 +254,7 @@ class AmazonConverseConfig(BaseConfig): @staticmethod def _convert_consecutive_user_messages_to_guarded_text( - messages: list[AllMessageValues], optional_params: dict + messages: list[AllMessageValues], optional_params: dict, keeps_cache_marks: bool = False ) -> list[AllMessageValues]: """ Convert consecutive user messages at the end to guarded_text type if guardrailConfig is present @@ -281,7 +292,11 @@ class AmazonConverseConfig(BaseConfig): new_content = [] for item in content: if isinstance(item, dict) and item.get("type") == "text": - new_item = {"type": "guarded_text", "text": item["text"]} + new_item = { + "type": "guarded_text", + "text": item["text"], + **_cache_mark_kept(_JSON_OBJECT.validate_python(item), keeps_cache_marks), + } new_content.append(new_item) else: new_content.append(item) @@ -289,7 +304,13 @@ class AmazonConverseConfig(BaseConfig): messages_copy[user_message_index]["content"] = new_content elif isinstance(content, str): # If content is a string, convert it to guarded_text - messages_copy[user_message_index]["content"] = [{"type": "guarded_text", "text": content}] + messages_copy[user_message_index]["content"] = [ + { + "type": "guarded_text", + "text": content, + **_cache_mark_kept(_JSON_OBJECT.validate_python(user_message), keeps_cache_marks), + } + ] return messages_copy @@ -1595,6 +1616,74 @@ class AmazonConverseConfig(BaseConfig): return ChatCompletionTextObject(type="text", text=text) return ChatCompletionTextObject(type="text", text=text, cache_control=cache_control) + def _messages_with_cache_marks( + self, + messages: Sequence[AllMessageValues], + model: str, + deployment_model: str, + drop_params: bool, + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], + model_info: Mapping[str, object] | None, + ) -> Sequence[AllMessageValues]: + if not bedrock_model_known_to_accept_cache_points(model, deployment_model, model_info): + return messages + markable: Final = ( + AnthropicCacheControlHook.prompt_cache_breakpoints_dropped_from_tool_results(messages, drop_params, model) + if "nova" in model.lower() + else messages + ) + return AnthropicCacheControlHook.prompt_cache_breakpoints_as_capped_cache_control( + markable, + drop_params, + AnthropicCacheControlHook.count_converse_cache_points, + lambda: self._tool_cache_points_on_the_wire( + model, deployment_model, optional_params, litellm_params, model_info + ), + ) + + def _tool_cache_points_on_the_wire( + self, + model: str, + deployment_model: str, + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], + model_info: Mapping[str, object] | None, + ) -> int: + tools: Final = optional_params.get("tools") + bedrock_tools, _ = self._process_tools_and_beta( + tools if isinstance(tools, list) else [], model, None, {}, litellm_params, deployment_model, model_info + ) + injects_tool_config_point: Final = bool(bedrock_tools) and any( + point.get("location") == "tool_config" + for point in configured_injection_points(optional_params.get("cache_control_injection_points")) + ) + return sum(1 for tool in bedrock_tools if "cachePoint" in tool) + int(injects_tool_config_point) + + def _guarded_text_keeps_cache_marks( + self, + messages: Sequence[AllMessageValues], + system_content_blocks: Sequence[SystemContentBlock], + model: str, + deployment_model: str, + optional_params: Mapping[str, object], + litellm_params: Mapping[str, object], + model_info: Mapping[str, object] | None, + ) -> bool: + if "guardrailConfig" not in optional_params or not bedrock_model_known_to_accept_cache_points( + model, deployment_model, model_info + ): + return False + message_cache_points: Final = sum(map(AnthropicCacheControlHook.count_converse_cache_points, messages)) + if message_cache_points == 0: + return False + cache_points: Final = ( + sum(1 for block in system_content_blocks if "cachePoint" in block) + + message_cache_points + + self._tool_cache_points_on_the_wire(model, deployment_model, optional_params, litellm_params, model_info) + ) + return cache_points <= MAX_CACHE_CONTROL_BLOCKS + def _transform_system_message( self, messages: list[AllMessageValues], @@ -2152,22 +2241,27 @@ class AmazonConverseConfig(BaseConfig): return data - @staticmethod def _claude_messages_within_cache_control_cap( + self, messages: Sequence[AllMessageValues], capability_model: str, + deployment_model: str, optional_params: Mapping[str, object], litellm_params: Mapping[str, object], + model_info: Mapping[str, object] | None, ) -> Sequence[AllMessageValues]: if not drops_unsupported_params(litellm_params.get("drop_params")) or "claude" not in capability_model.lower(): return messages from litellm.integrations.anthropic_cache_control_hook import ( - messages_within_cache_control_cap, + messages_within_reserved_cache_control_cap, tool_message_with_single_cache_mark, ) - return messages_within_cache_control_cap( - tuple(map(tool_message_with_single_cache_mark, messages)), optional_params.get("tools") + return messages_within_reserved_cache_control_cap( + tuple(map(tool_message_with_single_cache_mark, messages)), + self._tool_cache_points_on_the_wire( + capability_model, deployment_model, optional_params, litellm_params, model_info + ), ) async def _async_transform_request( @@ -2183,10 +2277,20 @@ class AmazonConverseConfig(BaseConfig): base_model: Final = litellm_params.get("base_model") capability_model: Final = bedrock_capability_model(model, base_model) model_info: Final = routed_model_info(litellm_params.get("model_info")) + json_optional_params: Final = _JSON_OBJECT.validate_python(optional_params) + marked_messages: Final = self._messages_with_cache_marks( + messages, + capability_model, + model, + drops_unsupported_params(litellm_params.get("drop_params")), + json_optional_params, + litellm_params, + model_info, + ) messages, system_content_blocks = self._transform_system_message( list( self._claude_messages_within_cache_control_cap( - messages, capability_model, _JSON_OBJECT.validate_python(optional_params), litellm_params + marked_messages, capability_model, model, json_optional_params, litellm_params, model_info ) ), model=capability_model, @@ -2195,8 +2299,19 @@ class AmazonConverseConfig(BaseConfig): ) # Convert last user message to guarded_text if guardrailConfig is present + guard_input: Final = self._handle_orphaned_tool_blocks(messages, optional_params) messages = self._convert_consecutive_user_messages_to_guarded_text( - self._handle_orphaned_tool_blocks(messages, optional_params), optional_params + guard_input, + optional_params, + keeps_cache_marks=self._guarded_text_keeps_cache_marks( + guard_input, + system_content_blocks, + capability_model, + model, + json_optional_params, + litellm_params, + model_info, + ), ) ## TRANSFORMATION ## @@ -2276,10 +2391,20 @@ class AmazonConverseConfig(BaseConfig): base_model: Final = litellm_params.get("base_model") capability_model: Final = bedrock_capability_model(model, base_model) model_info: Final = routed_model_info(litellm_params.get("model_info")) + json_optional_params: Final = _JSON_OBJECT.validate_python(optional_params) + marked_messages: Final = self._messages_with_cache_marks( + messages, + capability_model, + model, + drops_unsupported_params(litellm_params.get("drop_params")), + json_optional_params, + litellm_params, + model_info, + ) messages, system_content_blocks = self._transform_system_message( list( self._claude_messages_within_cache_control_cap( - messages, capability_model, _JSON_OBJECT.validate_python(optional_params), litellm_params + marked_messages, capability_model, model, json_optional_params, litellm_params, model_info ) ), model=capability_model, @@ -2288,8 +2413,19 @@ class AmazonConverseConfig(BaseConfig): ) # Convert last user message to guarded_text if guardrailConfig is present + guard_input: Final = self._handle_orphaned_tool_blocks(messages, optional_params) messages = self._convert_consecutive_user_messages_to_guarded_text( - self._handle_orphaned_tool_blocks(messages, optional_params), optional_params + guard_input, + optional_params, + keeps_cache_marks=self._guarded_text_keeps_cache_marks( + guard_input, + system_content_blocks, + capability_model, + model, + json_optional_params, + litellm_params, + model_info, + ), ) _data: Final[CommonRequestObject] = self._transform_request_helper( diff --git a/litellm/llms/bedrock/common_utils.py b/litellm/llms/bedrock/common_utils.py index 1f10b73505b..c1b3b5616b8 100644 --- a/litellm/llms/bedrock/common_utils.py +++ b/litellm/llms/bedrock/common_utils.py @@ -1559,22 +1559,63 @@ def bedrock_model_accepts_cache_points( def bedrock_declared_cache_point_support(model: str, model_info: Mapping[str, object] | None = None) -> bool | None: if _OPENAI_FAMILY_MODEL_RE.search(model): return False + explicit_marker_support: Final = _explicit_cache_point_support(model, model_info) + if explicit_marker_support is not None: + return explicit_marker_support + return _mapped_prompt_caching_support(model, model_info) + + +def bedrock_model_known_to_accept_cache_points( + model: str, deployment_model: str | None = None, model_info: Mapping[str, object] | None = None +) -> bool: + """Like ``bedrock_model_accepts_cache_points``, but only for the Claude and Nova families Bedrock documents + as taking cache checkpoints, unless the entry sets ``supports_prompt_cache_breakpoint``: an OpenAI + ``prompt_cache_breakpoint`` reaches every model of a mixed fleet, not only the ones a client marked for + Bedrock caching, and other families price implicit cache reads yet may reject the cachePoint field.""" + deployment_declared: Final = ( + None if deployment_model is None else _declared_marker_support(deployment_model, model_info) + ) + declared: Final = ( + deployment_declared if deployment_declared is not None else _declared_marker_support(model, model_info) + ) + return declared is True + + +def _declared_marker_support(model: str, model_info: Mapping[str, object] | None) -> bool | None: + if _OPENAI_FAMILY_MODEL_RE.search(model): + return False + explicit_marker_support: Final = _explicit_cache_point_support(model, model_info) + if explicit_marker_support is not None: + return explicit_marker_support + mapped: Final = _mapped_prompt_caching_support(model, model_info) + if not _documents_cache_checkpoints(model): + return None if mapped is None else False + return mapped is not False + + +def _documents_cache_checkpoints(model: str) -> bool: + return _is_claude(model) or get_bedrock_base_model(model).startswith("amazon.nova") + + +def _explicit_cache_point_support(model: str, model_info: Mapping[str, object] | None) -> bool | None: deployment_flag: Final = None if model_info is None else model_info.get("supports_prompt_cache_breakpoint") if deployment_flag is not None: return deployment_flag is True catalog_entry: Final = _catalog_entry_reader(model_info) - map_keys: Final = (model, get_bedrock_base_model(model)) registered_keys: Final = tuple(f"bedrock/{route}{model}" for route in ("", "converse/", "converse_like/")) - explicit_marker_support: Final = next( + return next( ( entry.get("supports_prompt_cache_breakpoint") is True - for key in (*registered_keys, *map_keys) + for key in (*registered_keys, model, get_bedrock_base_model(model)) if (entry := catalog_entry(key)) is not None and entry.get("supports_prompt_cache_breakpoint") is not None ), None, ) - if explicit_marker_support is not None: - return explicit_marker_support + + +def _mapped_prompt_caching_support(model: str, model_info: Mapping[str, object] | None) -> bool | None: + catalog_entry: Final = _catalog_entry_reader(model_info) + map_keys: Final = (model, get_bedrock_base_model(model)) if not any(catalog_entry(key) is not None for key in map_keys): return None deployment_caching: Final = None if model_info is None else lax_bool(model_info.get("supports_prompt_caching")) diff --git a/litellm/llms/opencode/chat/transformation.py b/litellm/llms/opencode/chat/transformation.py index 7e16f93d7cb..c6d288b6d84 100644 --- a/litellm/llms/opencode/chat/transformation.py +++ b/litellm/llms/opencode/chat/transformation.py @@ -112,6 +112,9 @@ class OpenCodeGoMessagesChatConfig(_GoReportedCostMixin, AnthropicConfig): def custom_llm_provider(self) -> str: return "opencode_go" + def translates_prompt_cache_breakpoint(self, model: str) -> bool: + return "claude" in model.lower() + def get_complete_url( self, api_base: str | None, diff --git a/litellm/router_utils/prompt_caching_cache.py b/litellm/router_utils/prompt_caching_cache.py index f78b5982186..1d91255e28c 100644 --- a/litellm/router_utils/prompt_caching_cache.py +++ b/litellm/router_utils/prompt_caching_cache.py @@ -55,6 +55,17 @@ def _sorted_pairs(pairs: Iterable[tuple[str, JsonValue]]) -> tuple[tuple[str, Js return tuple(sorted(pairs, key=lambda pair: pair[0])) +_CACHE_MARKER_KEYS: Final = frozenset(("cache_control", "prompt_cache_breakpoint")) + + +def _is_cache_marked(block: Mapping[str, object]) -> bool: + match block: + case {"cache_control": {"type": "ephemeral"}}: + return True + case _: + return block.get("prompt_cache_breakpoint") is not None + + def _canonical_bytes(value: object) -> bytes: return json.dumps(value, sort_keys=True, separators=(",", ":")).encode() @@ -66,7 +77,7 @@ def _block_unit( return _canonical_bytes((envelope, block)), message_run_type block_type: Final = block.get("type") block_run_type: Final = block_type if isinstance(block_type, str) and block_type in _TOOL_RUN_BLOCK_TYPES else None - stripped: Final = _sorted_pairs(item for item in block.items() if item[0] != "cache_control") + stripped: Final = _sorted_pairs(item for item in block.items() if item[0] not in _CACHE_MARKER_KEYS) return _canonical_bytes((envelope, stripped)), message_run_type or block_run_type @@ -264,15 +275,9 @@ class PromptCachingCache: continue for content_idx, content_block in enumerate(content): - if isinstance(content_block, dict): - cache_control = content_block.get("cache_control") - if ( - cache_control is not None - and isinstance(cache_control, dict) - and cache_control.get("type") == "ephemeral" - ): - last_cacheable_message_idx = msg_idx - last_cacheable_content_idx = content_idx + if isinstance(content_block, dict) and _is_cache_marked(content_block): + last_cacheable_message_idx = msg_idx + last_cacheable_content_idx = content_idx # If no cacheable block found, return empty list (no cacheable prefix) if last_cacheable_message_idx is None: diff --git a/tests/unit/integrations/test_anthropic_cache_control_hook.py b/tests/unit/integrations/test_anthropic_cache_control_hook.py index c040d8e5bac..766038d5c4f 100644 --- a/tests/unit/integrations/test_anthropic_cache_control_hook.py +++ b/tests/unit/integrations/test_anthropic_cache_control_hook.py @@ -1401,6 +1401,109 @@ def _marks_the_route_sends(model: str, custom_llm_provider: str | None, messages _SECOND_SEARCH_CALL: Final = {**_SEARCH_CALL, "id": "srvtoolu_02Search"} +_CLIENT_MARKED_CONVERSATION: Final = ( + {"role": "system", "content": [{"type": "text", "text": "Handbook", "prompt_cache_breakpoint": {"mode": "explicit"}}]}, + {"role": "user", "content": [{"type": "text", "text": "First", "prompt_cache_breakpoint": {"mode": "explicit"}}]}, + {"role": "assistant", "content": "First answer"}, + {"role": "user", "content": "Second"}, +) + + +def _hook_result( + model: str, points: tuple[Mapping[str, object], ...] = ({"location": "message", "index": -1},), **params: object +) -> tuple[list[AllMessageValues], Mapping[str, object]]: + _, processed, returned_params = AnthropicCacheControlHook().get_chat_completion_prompt( + model=model, + messages=cast(list[AllMessageValues], copy.deepcopy(list(_CLIENT_MARKED_CONVERSATION))), + non_default_params={"cache_control_injection_points": list(points), **params}, + prompt_id=None, + prompt_variables=None, + dynamic_callback_params={}, + ) + return processed, returned_params + + +def _hook_output( + model: str, points: tuple[Mapping[str, object], ...] = ({"location": "message", "index": -1},), **params: object +) -> list[AllMessageValues]: + return _hook_result(model, points, **params)[0] + + +def test_cache_control_injection_points_replace_client_prompt_cache_breakpoints_on_claude(): + request: Final = AnthropicConfig().transform_request( + model="claude-sonnet-5-5", + messages=_hook_output("anthropic/claude-sonnet-5-5"), + optional_params={"max_tokens": 64}, + litellm_params={}, + headers={}, + ) + + assert _marks_in(request["system"]) == 0 + assert [_marks_in(message) for message in request["messages"]] == [0, 0, 1] + assert "prompt_cache_breakpoint" not in json.dumps(request) + + +@pytest.mark.parametrize( + "sibling_model_infos", + [ + pytest.param((), id="one-deployment"), + pytest.param( + ({"supports_prompt_cache_breakpoint": True, "base_model": "anthropic.claude-unmapped-v1:0"},), + id="beside-a-deployment-of-the-same-profile-with-a-base-model", + ), + ], +) +@pytest.mark.parametrize("route", ["bedrock", "bedrock/converse"]) +def test_cache_control_injection_points_replace_client_prompt_cache_breakpoints_on_a_flagged_bedrock_profile( + local_model_cost_map, sibling_model_infos: tuple[Mapping[str, object], ...], route: str +): + from litellm import Router + + model: Final = f"{route}/arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/claude-flagged" + Router( + model_list=[ + { + "model_name": f"claude-profile-{position}", + "litellm_params": {"model": model, "aws_region_name": "us-east-1"}, + "model_info": model_info, + } + for position, model_info in enumerate(({"supports_prompt_cache_breakpoint": True}, *sibling_model_infos)) + ] + ) + processed, params = _hook_result(model) + + assert "prompt_cache_breakpoint" not in json.dumps(processed) + assert "prompt_cache_options" not in params + assert [_marks_in(message) for message in processed] == [0, 0, 0, 1] + + +def test_tool_config_injection_point_leaves_client_prompt_cache_breakpoints_on_claude(): + processed: Final = _hook_output("anthropic/claude-sonnet-5-5", points=({"location": "tool_config"},)) + + assert [message["content"] for message in processed[:2]] == [ + message["content"] for message in _CLIENT_MARKED_CONVERSATION[:2] + ] + + +@pytest.mark.parametrize( + ("model", "params"), + [ + pytest.param("openai/gpt-5.6", {"api_base": "https://gateway.example.com/v1"}, id="reads-them"), + pytest.param("openai/gpt-4o", {}, id="openai-model-that-predates-them"), + pytest.param("hosted_vllm/qwen3", {"api_base": "http://vllm.example.com/v1"}, id="self-hosted"), + pytest.param("azure_ai/qa-unflagged-gpt", {"api_base": "https://qa.services.ai.azure.com/models"}, id="foundry"), + ], +) +def test_cache_control_injection_points_keep_client_prompt_cache_breakpoints_on_a_route_that_does_not_translate_them( + model: str, params: Mapping[str, object] +): + processed: Final = _hook_output(model, **params) + + assert [message["content"] for message in processed[:2]] == [ + message["content"] for message in _CLIENT_MARKED_CONVERSATION[:2] + ] + + @pytest.mark.parametrize( ("model", "custom_llm_provider"), [ @@ -4965,6 +5068,33 @@ class TestMalformedInjectionPointsAreIgnored: assert kwargs == {"cache_control_injection_points": value} +def test_capping_thousands_of_marks_on_one_tool_result_counts_the_request_a_logarithmic_number_of_times(): + def marked_parts(prefix: str, count: int) -> list[dict]: + return [ + {"type": "text", "text": f"{prefix} {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(count) + ] + + messages: Final = [ + {"role": "user", "content": marked_parts("part", 4)}, + { + "role": "assistant", + "content": None, + "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": marked_parts("result", 2000)}, + ] + sent_breakpoints: Final = MagicMock(side_effect=AnthropicCacheControlHook.count_converse_cache_points) + + capped: Final = AnthropicCacheControlHook.prompt_cache_breakpoints_as_capped_cache_control( + cast(List[AllMessageValues], messages), True, sent_breakpoints, lambda: 0 + ) + + assert [part.get("cache_control") for part in capped[0]["content"]] == [{"type": "ephemeral"}] * 3 + [None] + assert ["cache_control" in part for part in capped[2]["content"]] == [False] * 1997 + [True] * 3 + assert sent_breakpoints.call_count < 3 * 20 + + def test_cache_control_hook_leaves_the_duplicate_tool_result_warning_to_the_request_builder( monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture ): diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py index 13ff11e8a3f..fc873832956 100644 --- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py +++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py @@ -1,4 +1,5 @@ import copy +import itertools import json import logging from collections.abc import Iterator @@ -4422,6 +4423,184 @@ def test_transform_request_includes_cache_control(): assert result["cache_control"] == {"type": "ephemeral"} +def _anthropic_body(messages: list, litellm_params: dict | None = None, tools: list | None = None) -> dict: + return AnthropicConfig().transform_request( + model="claude-sonnet-5-5", + messages=messages, + optional_params={"max_tokens": 100, **({"tools": tools} if tools else {})}, + litellm_params=litellm_params or {}, + headers={}, + ) + + +def test_transform_request_carries_prompt_cache_breakpoint_to_cache_control(): + body: Final = _anthropic_body( + [ + { + "role": "system", + "content": [ + {"type": "text", "text": "You are terse.", "prompt_cache_breakpoint": {"mode": "explicit"}} + ], + }, + { + "role": "user", + "content": [ + {"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit"}}, + { + "type": "text", + "text": "Bye", + "cache_control": {"type": "ephemeral", "ttl": "1h"}, + "prompt_cache_breakpoint": {"mode": "explicit"}, + }, + ], + }, + ] + ) + + assert body["system"] == [{"type": "text", "text": "You are terse.", "cache_control": {"type": "ephemeral"}}] + assert body["messages"] == [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Hello", "cache_control": {"type": "ephemeral"}}, + {"type": "text", "text": "Bye", "cache_control": {"type": "ephemeral", "ttl": "1h"}}, + ], + } + ] + + +@pytest.mark.parametrize( + "marker, expected_block", + [ + ({"mode": "explicit", "ttl": "30m"}, {"type": "text", "text": "Hello", "cache_control": {"type": "ephemeral"}}), + ({"mode": "implicit"}, {"type": "text", "text": "Hello"}), + ], +) +def test_transform_request_prompt_cache_breakpoint_without_claude_equivalent(marker, expected_block): + messages: Final = [{"role": "user", "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": marker}]}] + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="drop_params"): + _anthropic_body(copy.deepcopy(messages)) + + body: Final = _anthropic_body(messages, litellm_params={"drop_params": True}) + assert body["messages"] == [{"role": "user", "content": [expected_block]}] + + +def test_transform_request_names_the_marker_ttl_rule_for_a_ttl_only_claude_takes(): + messages: Final = [ + {"role": "user", "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}}]} + ] + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="optional ttl of '30m' or '12h'"): + _anthropic_body(messages) + + +def test_transform_request_keeps_the_first_and_last_prompt_cache_breakpoints_within_the_cap(): + client_marked_part: Final = {"type": "text", "text": "part 0", "cache_control": {"type": "ephemeral"}} + marked_parts: Final = [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(1, 5) + ] + messages: Final = [{"role": "user", "content": [client_marked_part, *marked_parts]}] + tools: Final = [{"name": "lookup", "input_schema": {"type": "object"}, "cache_control": {"type": "ephemeral"}}] + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="at most 4 cache breakpoints"): + _anthropic_body(copy.deepcopy(messages), tools=copy.deepcopy(tools)) + + body: Final = _anthropic_body(messages, litellm_params={"drop_params": True}, tools=tools) + assert [part.get("cache_control") for part in body["messages"][0]["content"]] == [ + {"type": "ephemeral"}, + {"type": "ephemeral"}, + None, + None, + {"type": "ephemeral"}, + ] + assert body["tools"][0]["cache_control"] == {"type": "ephemeral"} + + +def test_transform_request_counts_a_cache_marked_tool_search_tool_toward_the_cap(): + config: Final = AnthropicConfig() + optional_params: Final = config.map_openai_params( + non_default_params={ + "max_tokens": 100, + "tools": [ + { + "type": "tool_search_tool_regex_20251119", + "name": "tool_search_tool_regex", + "cache_control": {"type": "ephemeral"}, + } + ], + }, + optional_params={}, + model="claude-sonnet-5-5", + drop_params=False, + ) + marked_turns: Final = [ + { + "role": "user", + "content": [{"type": "text", "text": f"turn {index}", "prompt_cache_breakpoint": {"mode": "explicit"}}], + } + for index in range(4) + ] + + def body(litellm_params: dict) -> dict: + return config.transform_request( + model="claude-sonnet-5-5", + messages=copy.deepcopy(marked_turns), + optional_params=copy.deepcopy(optional_params), + litellm_params=litellm_params, + headers={}, + ) + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="this request has 5"): + body({}) + + capped: Final = body({"drop_params": True}) + assert capped["tools"][0]["cache_control"] == {"type": "ephemeral"} + sent_parts: Final = itertools.chain.from_iterable(message["content"] for message in capped["messages"]) + assert [part.get("cache_control") for part in sent_parts] == [ + {"type": "ephemeral"}, + {"type": "ephemeral"}, + None, + {"type": "ephemeral"}, + ] + + +@pytest.mark.parametrize( + "unsent_marks", + [ + pytest.param( + [ + { + "role": "assistant", + "content": [{"type": "text", "text": "Noted.", "prompt_cache_breakpoint": {"mode": "explicit"}}], + } + ], + id="assistant_part", + ), + pytest.param( + [{"role": "user", "content": "Also this.", "prompt_cache_breakpoint": {"mode": "explicit"}}], + id="message_level", + ), + ], +) +def test_transform_request_caps_only_the_marks_claude_receives(unsent_marks: list[dict]): + marked_parts: Final = [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} for index in range(4) + ] + + body: Final = _anthropic_body( + [{"role": "user", "content": marked_parts}, *unsent_marks, {"role": "user", "content": "Go on."}] + ) + + sent_marks: Final = [ + part.get("cache_control") + for part in itertools.chain.from_iterable(message["content"] for message in body["messages"]) + if isinstance(part, dict) + ] + assert sent_marks.count({"type": "ephemeral"}) == 4 + + def test_compaction_block_empty_list_not_added(): """ Test that empty compaction_blocks list is not added to provider_specific_fields. diff --git a/tests/unit/llms/bedrock/chat/test_converse_transformation.py b/tests/unit/llms/bedrock/chat/test_converse_transformation.py index 8cab1018d25..7c19ef5b751 100644 --- a/tests/unit/llms/bedrock/chat/test_converse_transformation.py +++ b/tests/unit/llms/bedrock/chat/test_converse_transformation.py @@ -10073,6 +10073,461 @@ def test_bedrock_prompt_caching_message(messages, expected_cache_control): assert "cachePoint" not in json.dumps(transformed_messages) +def test_converse_carries_prompt_cache_breakpoint_to_cache_point(): + request: Final = litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=[ + { + "role": "system", + "content": [ + {"type": "text", "text": "You are terse.", "prompt_cache_breakpoint": {"mode": "explicit"}} + ], + }, + { + "role": "user", + "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit"}}], + }, + ], + optional_params={}, + litellm_params={}, + ) + + assert request["system"] == [{"text": "You are terse."}, {"cachePoint": {"type": "default"}}] + assert request["messages"] == [ + {"role": "user", "content": [{"text": "Hello"}, {"cachePoint": {"type": "default"}}]} + ] + + +def test_converse_keeps_the_first_and_last_prompt_cache_breakpoints_within_the_cap(): + messages: Final = [ + { + "role": "user", + "content": [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(5) + ], + } + ] + + def transform(litellm_params: dict) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=copy.deepcopy(messages), + optional_params={}, + litellm_params=litellm_params, + ) + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="at most 4 cache breakpoints"): + transform({}) + + cache_point: Final = {"cachePoint": {"type": "default"}} + assert transform({"drop_params": True})["messages"] == [ + { + "role": "user", + "content": [ + {"text": "part 0"}, + cache_point, + {"text": "part 1"}, + cache_point, + {"text": "part 2"}, + {"text": "part 3"}, + cache_point, + {"text": "part 4"}, + cache_point, + ], + } + ] + + +_CONVERSE_CAP_TOOL: Final[dict[str, object]] = { + "type": "function", + "function": {"name": "lookup", "parameters": {"type": "object", "properties": {}}}, +} + + +@pytest.mark.parametrize( + "tool_cache_params", + [ + {"tools": [{**_CONVERSE_CAP_TOOL, "cache_control": {"type": "ephemeral"}}]}, + { + "tools": [_CONVERSE_CAP_TOOL], + "cache_control_injection_points": [{"location": "tool_config", "control": {"type": "ephemeral"}}], + }, + ], + ids=["client_tool_cache_control", "tool_config_injection_point"], +) +def test_converse_reserves_the_tool_cache_point_within_the_cap(tool_cache_params: dict[str, object]): + messages: Final = [ + { + "role": "user", + "content": [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(4) + ], + } + ] + + def transform(litellm_params: dict) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=copy.deepcopy(messages), + optional_params=copy.deepcopy(tool_cache_params), + litellm_params=litellm_params, + ) + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="this request has 5"): + transform({}) + + request: Final = transform({"drop_params": True}) + cache_point: Final = {"cachePoint": {"type": "default"}} + assert request["messages"][0]["content"] == [ + {"text": "part 0"}, + cache_point, + {"text": "part 1"}, + cache_point, + {"text": "part 2"}, + {"text": "part 3"}, + cache_point, + ] + assert request["toolConfig"]["tools"][-1] == cache_point + + + +@pytest.mark.parametrize( + "unsent_tool_params", + [ + { + "tools": [ + _CONVERSE_CAP_TOOL, + {"type": "tool_search_tool_regex_20251119", "name": "find", "cache_control": {"type": "ephemeral"}}, + ] + }, + {"tools": [_CONVERSE_CAP_TOOL, {"type": "web_search", "cache_control": {"type": "ephemeral"}}]}, + { + "tools": [{"type": "web_search"}], + "cache_control_injection_points": [{"location": "tool_config", "control": {"type": "ephemeral"}}], + }, + ], + ids=["tool_search_tool", "built_in_tool", "tool_config_point_with_no_tool_sent"], +) +@pytest.mark.parametrize("litellm_params", [{}, {"drop_params": True}], ids=["strict", "drop_params"]) +def test_converse_leaves_the_cap_to_markers_when_a_marked_tool_is_not_sent( + unsent_tool_params: dict[str, object], litellm_params: dict[str, object] +): + request: Final = litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(4) + ], + } + ], + optional_params=copy.deepcopy(unsent_tool_params), + litellm_params=dict(litellm_params), + ) + + cache_point: Final = {"cachePoint": {"type": "default"}} + assert request["messages"][0]["content"] == [ + {"text": "part 0"}, + cache_point, + {"text": "part 1"}, + cache_point, + {"text": "part 2"}, + cache_point, + {"text": "part 3"}, + cache_point, + ] + assert cache_point not in request.get("toolConfig", {}).get("tools", []) + +def _converse_tool_turn(*result_parts: dict[str, object]) -> list[dict[str, object]]: + return [ + { + "role": "assistant", + "content": None, + "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}], + }, + {"role": "tool", "tool_call_id": "call_1", "content": list(result_parts)}, + ] + + +def _marked_text_parts(count: int) -> list[dict[str, object]]: + return [ + {"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(count) + ] + + +def _marked_tool_result_parts() -> list[dict[str, object]]: + return [ + {"type": "text", "text": f"result {index}", "prompt_cache_breakpoint": {"mode": "explicit"}} + for index in range(2) + ] + + +def test_converse_counts_one_cache_point_for_a_tool_result_with_several_marks(): + request: Final = litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=[ + {"role": "user", "content": _marked_text_parts(3)}, + *_converse_tool_turn(*_marked_tool_result_parts()), + ], + optional_params={"tools": [_CONVERSE_CAP_TOOL]}, + litellm_params={}, + ) + + assert len(_collect_cache_points(request["messages"])) == 4 + + +def test_converse_drops_marks_until_a_merged_tool_result_fits_the_cap(): + messages: Final = [ + {"role": "user", "content": _marked_text_parts(4)}, + *_converse_tool_turn(*_marked_tool_result_parts()), + ] + + def transform(litellm_params: dict) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=copy.deepcopy(messages), + optional_params={"tools": [_CONVERSE_CAP_TOOL]}, + litellm_params=litellm_params, + ) + + with pytest.raises(litellm.UnsupportedParamsError, match="this request has 5"): + transform({}) + + request: Final = transform({"drop_params": True}) + cache_point: Final = {"cachePoint": {"type": "default"}} + assert request["messages"][0]["content"] == [ + {"text": "part 0"}, + cache_point, + {"text": "part 1"}, + cache_point, + {"text": "part 2"}, + cache_point, + {"text": "part 3"}, + ] + assert len(_collect_cache_points(request["messages"][1:])) == 1 + + +def test_converse_drops_prompt_cache_breakpoint_from_a_nova_tool_result(): + messages: Final = [ + { + "role": "system", + "content": [{"type": "text", "text": "You are terse.", "prompt_cache_breakpoint": {"mode": "explicit"}}], + }, + {"role": "user", "content": "Look it up."}, + *_converse_tool_turn(*_marked_tool_result_parts()), + ] + + def transform(litellm_params: dict) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model="us.amazon.nova-lite-v1:0", + messages=copy.deepcopy(messages), + optional_params={"tools": [_CONVERSE_CAP_TOOL]}, + litellm_params=litellm_params, + ) + + with pytest.raises(litellm.UnsupportedParamsError, match="tool result"): + transform({}) + + request: Final = transform({"drop_params": True}) + assert request["system"] == [{"text": "You are terse."}, {"cachePoint": {"type": "default"}}] + assert _collect_cache_points(request["messages"]) == [] + + +def test_converse_translates_prompt_cache_breakpoint_on_an_unmapped_arn_only_when_flagged(local_model_cost_map): + from litellm import Router + + flagged_arn: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/claude-flagged" + unflagged_arn: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/mistral" + Router( + model_list=[ + { + "model_name": "claude-profile", + "litellm_params": {"model": f"bedrock/{flagged_arn}", "aws_region_name": "us-east-1"}, + "model_info": {"supports_prompt_cache_breakpoint": True}, + }, + { + "model_name": "mistral-profile", + "litellm_params": {"model": f"bedrock/{unflagged_arn}", "aws_region_name": "us-east-1"}, + }, + ] + ) + + def user_content(model: str, ttl: dict) -> list: + return litellm.AmazonConverseConfig()._transform_request( + model=model, + messages=[ + { + "role": "user", + "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit", **ttl}}], + } + ], + optional_params={}, + litellm_params={}, + )["messages"][0]["content"] + + assert user_content(flagged_arn, {}) == [{"text": "Hello"}, {"cachePoint": {"type": "default"}}] + assert user_content(unflagged_arn, {}) == [{"text": "Hello"}] + assert user_content(unflagged_arn, {"ttl": "30m"}) == [{"text": "Hello"}] + + + +def test_converse_prompt_cache_breakpoint_follows_the_arn_flag_over_its_base_model(local_model_cost_map): + from litellm import Router + + opted_in_arn: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/claude-next" + opted_out_arn: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/claude-opted-out" + Router( + model_list=[ + { + "model_name": "opted-in", + "litellm_params": {"model": f"bedrock/{opted_in_arn}", "aws_region_name": "us-east-1"}, + "model_info": {"supports_prompt_cache_breakpoint": True}, + }, + { + "model_name": "opted-out", + "litellm_params": {"model": f"bedrock/{opted_out_arn}", "aws_region_name": "us-east-1"}, + "model_info": {"supports_prompt_cache_breakpoint": False}, + }, + ] + ) + + def user_content(model: str, base_model: str, ttl: dict) -> list: + return litellm.AmazonConverseConfig()._transform_request( + model=model, + messages=[ + { + "role": "user", + "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit", **ttl}}], + } + ], + optional_params={}, + litellm_params={"base_model": base_model}, + )["messages"][0]["content"] + + assert user_content(opted_in_arn, "anthropic.claude-unmapped-v1:0", {}) == [ + {"text": "Hello"}, + {"cachePoint": {"type": "default"}}, + ] + assert user_content(opted_out_arn, "anthropic.claude-sonnet-5-5", {"ttl": "30m"}) == [{"text": "Hello"}] + + +def test_converse_translates_prompt_cache_breakpoint_only_for_families_that_take_cache_points(local_model_cost_map): + litellm.register_model( + {"zai.glm-next-v1:0": {"litellm_provider": "bedrock_converse", "mode": "chat", "supports_prompt_caching": True}} + ) + + def user_content(model: str, mark: dict) -> list: + return litellm.AmazonConverseConfig()._transform_request( + model=model, + messages=[{"role": "user", "content": [{"type": "text", "text": "Hello", **mark}]}], + optional_params={}, + litellm_params={}, + )["messages"][0]["content"] + + marker: Final = {"prompt_cache_breakpoint": {"mode": "explicit"}} + client_mark: Final = {"cache_control": {"type": "ephemeral"}} + cache_point: Final = {"cachePoint": {"type": "default"}} + assert user_content("zai.glm-next-v1:0", marker) == [{"text": "Hello"}] + assert user_content("zai.glm-next-v1:0", client_mark) == [{"text": "Hello"}, cache_point] + assert user_content("us.anthropic.claude-next-v1:0", marker) == [{"text": "Hello"}, cache_point] + assert user_content("us.amazon.nova-next-v1:0", marker) == [{"text": "Hello"}, cache_point] + + +def test_converse_prompt_cache_breakpoint_on_guarded_user_text_keeps_its_cache_point(local_model_cost_map): + guardrail: Final = {"guardrailConfig": {"guardrailIdentifier": "gr-1", "guardrailVersion": "DRAFT"}} + + def request(system_marks: int) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=[ + *( + { + "role": "system", + "content": [{"type": "text", "text": f"Rule {i}", "cache_control": {"type": "ephemeral"}}], + } + for i in range(system_marks) + ), + { + "role": "user", + "content": [{"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit"}}], + }, + ], + optional_params=dict(guardrail), + litellm_params={}, + ) + + assert request(0)["messages"][0]["content"] == [ + {"guardContent": {"text": {"text": "Hello"}}}, + {"cachePoint": {"type": "default"}}, + ] + with pytest.raises(litellm.UnsupportedParamsError): + request(4) + + client_marked: Final = litellm.AmazonConverseConfig()._transform_request( + model="anthropic.claude-sonnet-5-5", + messages=[ + *( + { + "role": "system", + "content": [{"type": "text", "text": f"Rule {i}", "cache_control": {"type": "ephemeral"}}], + } + for i in range(4) + ), + {"role": "user", "content": [{"type": "text", "text": "Hello", "cache_control": {"type": "ephemeral"}}]}, + ], + optional_params=dict(guardrail), + litellm_params={}, + ) + assert sum("cachePoint" in block for block in client_marked["system"]) == 4 + assert client_marked["messages"][0]["content"] == [{"guardContent": {"text": {"text": "Hello"}}}] + + +def test_converse_guarded_user_text_keeps_no_cache_point_on_a_model_not_known_to_take_one(local_model_cost_map): + request: Final = litellm.AmazonConverseConfig()._transform_request( + model="arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/unmapped1", + messages=[{"role": "user", "content": [{"type": "text", "text": "Hello", "cache_control": {"type": "ephemeral"}}]}], + optional_params={"guardrailConfig": {"guardrailIdentifier": "gr-1", "guardrailVersion": "DRAFT"}}, + litellm_params={}, + ) + + assert request["messages"][0]["content"] == [{"guardContent": {"text": {"text": "Hello"}}}] + + +def _converse_request_with_30m_breakpoint(model: str, litellm_params: dict) -> dict: + return litellm.AmazonConverseConfig()._transform_request( + model=model, + messages=[ + { + "role": "user", + "content": [ + {"type": "text", "text": "Hello", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}} + ], + } + ], + optional_params={}, + litellm_params=litellm_params, + ) + + +def test_converse_rejects_30m_breakpoint_ttl_only_where_a_cache_point_is_written(): + claude: Final = "anthropic.claude-sonnet-5-5" + + with pytest.raises(litellm.utils.UnsupportedParamsError, match="drop_params"): + _converse_request_with_30m_breakpoint(claude, {}) + + assert _converse_request_with_30m_breakpoint(claude, {"drop_params": True})["messages"] == [ + {"role": "user", "content": [{"text": "Hello"}, {"cachePoint": {"type": "default"}}]} + ] + assert _converse_request_with_30m_breakpoint("meta.llama3-1-8b-instruct-v1:0", {})["messages"] == [ + {"role": "user", "content": [{"text": "Hello"}]} + ] + + @pytest.mark.parametrize( "model, expected_supports_tool_call", [ diff --git a/tests/unit/llms/opencode/chat/test_transformation.py b/tests/unit/llms/opencode/chat/test_transformation.py index 91361a3d81d..08c80db294c 100644 --- a/tests/unit/llms/opencode/chat/test_transformation.py +++ b/tests/unit/llms/opencode/chat/test_transformation.py @@ -14,6 +14,7 @@ from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.core_helpers import RESPONSE_COST_HEADER from litellm.litellm_core_utils.litellm_logging import Logging from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.llms.opencode.chat.transformation import OpenCodeGoMessagesChatConfig from litellm.llms.opencode.common_utils import go_headers @@ -591,3 +592,22 @@ def test_stream_chunk_builder_reads_list_cost_from_dict_chunks(): assert assembled is not None assert assembled._hidden_params["response_cost"] == GO_REPORTED_USAGE["opencode"]["list_cost"] + + +@pytest.mark.parametrize(("model", "cache_marks"), [("qwen3.8-flash", 0), ("minimax-m3", 0), ("claude-haiku-5-5", 4)]) +def test_go_messages_route_carries_prompt_cache_breakpoint_to_cache_control_only_for_claude(model: str, cache_marks: int): + marked_notes: Final = [ + {"type": "text", "text": f"Note {n}", "prompt_cache_breakpoint": {"mode": "explicit"}} for n in range(4) + ] + + request: Final = OpenCodeGoMessagesChatConfig().transform_request( + model=model, + messages=[{"role": "user", "content": [*marked_notes, {"type": "text", "text": "Answer in one line."}]}], + optional_params={"max_tokens": 64}, + litellm_params={}, + headers={}, + ) + + wire: Final = json.dumps(request) + assert wire.count('"cache_control"') == cache_marks + assert "prompt_cache_breakpoint" not in wire diff --git a/tests/unit/router_utils/test_prompt_caching_cache.py b/tests/unit/router_utils/test_prompt_caching_cache.py index ce8f04b8c54..1500d7433d2 100644 --- a/tests/unit/router_utils/test_prompt_caching_cache.py +++ b/tests/unit/router_utils/test_prompt_caching_cache.py @@ -1,4 +1,5 @@ import asyncio +from collections.abc import Mapping from typing import Final, cast import pytest @@ -377,6 +378,32 @@ def test_extract_cacheable_prefix_mixed_string_and_list_content(): assert isinstance(result[2]["content"], list) +def test_prompt_cache_breakpoint_marks_the_same_cacheable_prefix_as_cache_control(): + def _messages(marker: Mapping[str, Mapping[str, str]]) -> list[AllMessageValues]: + return cast( # cast-ok: the text part type does not declare the prompt_cache_breakpoint key + list[AllMessageValues], + [ + { + "role": "system", + "content": [ + {"type": "text", "text": "Here is the full text of a complex legal agreement" * 400, **marker} + ], + }, + {"role": "user", "content": "what are the key terms?"}, + ], + ) + + marked_key: Final = PromptCachingCache.get_prompt_caching_cache_key( + _messages({"prompt_cache_breakpoint": {"mode": "explicit"}}), None + ) + + assert marked_key is not None + assert marked_key == PromptCachingCache.get_prompt_caching_cache_key( + _messages({"cache_control": {"type": "ephemeral"}}), None + ) + assert PromptCachingCache.get_prompt_caching_cache_key(_messages({}), None) is None + + @pytest.mark.asyncio async def test_same_cacheable_prefix_routes_to_same_deployment() -> None: from litellm import Router