scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-11 08:38:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 280

fix(anthropic)!: carry prompt_cache_breakpoint to cache_control on chat completions to Claude (#49783)

Browse source 
* fix(anthropic): carry prompt_cache_breakpoint to cache_control on chat completions to Claude

A content part carrying OpenAI's prompt_cache_breakpoint on a /v1/chat/completions request to Claude (Anthropic, Bedrock Invoke and Converse, Vertex AI, Azure AI) now reaches Claude as cache_control {"type": "ephemeral"} on the same block, so prompt caching works for OpenAI-shaped clients instead of the marker being stripped.

Only parts OpenAI accepts the marker on are translated, a client cache_control on the same block wins, and Converse translates only for models that take cache points. A 30m ttl or a malformed marker answers UnsupportedParamsError without drop_params and becomes a default 5m mark or no mark with it.

* fix(anthropic): hold translated prompt_cache_breakpoint marks to the 4-breakpoint cap

Marks past Claude's 4 cache breakpoints now answer 400 without drop_params and
keep the earliest under it, counting the client's own message, tool, and
top-level cache_control marks. The ttl and malformed-marker errors name
drop_params for proxy users too and no longer name a single provider.

* fix(bedrock): count Converse tool cache points in the breakpoint cap and gate unmapped models

A client tool cache_control or a tool_config injection point becomes a Converse tool
cachePoint, so it now takes one of the 4 slots before a translated marker does.

Converse only translates prompt_cache_breakpoint for a model the cost map says takes
prompt caching. An unmapped model such as an application inference profile ARN opts in
through model_info supports_prompt_cache_breakpoint, so a marker sent to every model of a
mixed fleet no longer turns a non-caching profile's 200 into Bedrock's 403.

Router prompt-caching affinity now treats a marked block like a cache_control one, and
requests without a marker skip the translation after one pass over their messages.

* fix(bedrock): count only the tool cache points Converse sends against the breakpoint cap

* fix(anthropic,bedrock): cap only the cache breakpoints each provider sends, and drop marks Nova cannot take on tool results

* perf(anthropic,bedrock): find the breakpoint cut-off by binary search so trimming many marks stays fast

* fix(anthropic): drop the duplicate drop_params left by the main merge

* fix(bedrock): let an inference profile's cache breakpoint flag win over its base_model

* fix(anthropic): keep the first and last prompt_cache_breakpoint marks past the cap

Under drop_params the translated marks that fit the cap are now split between
the start and the end of the request, so the stable prefix and the newest turns
both stay cached. The cap count also includes a native tool search tool that
keeps its cache_control on the wire.

Bedrock Converse now translates the marker only for the Claude and Nova
families, which Bedrock documents as taking cache checkpoints, unless the
deployment sets supports_prompt_cache_breakpoint. A guarded user turn keeps its
cache mark when the request fits the cap.

* refactor(bedrock): decide guarded cache marks with a typed bool helper

* fix(bedrock): validate guarded blocks before reading their cache mark

* fix(anthropic): let injection points replace client marks and keep non-Claude routes untranslated

When a deployment sets cache_control_injection_points, the client's
prompt_cache_breakpoint marks are dropped on targets that do not read them, so
the operator's injection points decide every Claude breakpoint.

OpenCode Go models served over the Anthropic messages endpoint translate the
marker only for Claude, and guarded Converse text keeps a cache mark only on a
model known to take cache points.

* fix(anthropic): keep client marks unless a message injection point replaces them on a translating route

A tool_config-only injection point no longer strips client prompt_cache_breakpoint marks, a flagged Bedrock profile no longer counts as reading the marker natively, and guarded Converse text skips the tool count when no message carries a cache mark.

* fix(anthropic): keep Bedrock and Claude deployments flagged for the marker on cache_control injection

A supports_prompt_cache_breakpoint flag on a Bedrock profile or a Claude deployment made the injection hook write OpenAI markers and prompt_cache_options, which Bedrock answers 400 for. Routes that translate the marker now never take the OpenAI dialect.

* fix(bedrock): reserve only the tool cache points Converse sends when trimming Claude cache marks under drop_params

* fix(anthropic): name the marker's accepted ttl values in the malformed prompt_cache_breakpoint 400

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
Co-authored-by: ryan-crabbe-berri <ryan@berri.ai>
This commit is contained in:
zoveki 2026-11-03 04:56:22 -08:00 • committed by GitHub
parent ff182c194e
commit baddc67041
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1308 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

228
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -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:

50
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -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])

20
litellm/llms/anthropic/chat/transformation.py
View file

@ -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):

158
litellm/llms/bedrock/chat/converse_transformation.py
View file

@ -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(

51
litellm/llms/bedrock/common_utils.py
View file

@ -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"))

3
litellm/llms/opencode/chat/transformation.py
View file

@ -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,

25
litellm/router_utils/prompt_caching_cache.py
View file

@ -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:

130
tests/unit/integrations/test_anthropic_cache_control_hook.py
View file

@ -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
):

179
tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py
View file

@ -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.

455
tests/unit/llms/bedrock/chat/test_converse_transformation.py
View file

@ -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",
[

20
tests/unit/llms/opencode/chat/test_transformation.py
View file

@ -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

27
tests/unit/router_utils/test_prompt_caching_cache.py
View file

@ -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