scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

refactor(anthropic): translate prompt_cache_breakpoint and prompt_cache_options for Claude once in completion()

Browse source 
Replaces the per-transform census with one step in completion() for every Claude chat route.
Breakpoints become cache_control on the same block. A ttl Claude lacks is a 400 naming drop_params,
or dropped with the mark kept under drop_params. prompt_cache_options with mode implicit turns on
Claude automatic caching where the route has it and is a 400 on Converse. With drop_params, more
than four marks keep the latest four. Without it they reach Claude, which answers itself
This commit is contained in:
ryan-crabbe-berri 2026-10-14 08:25:30 -07:00
parent fedb3f43cf
commit a5b488f38c
17 changed files with 527 additions and 2337 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -7,7 +7,6 @@ import json
import mimetypes
import re
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from collections.abc import Set as AbstractSet
from itertools import groupby, islice
from os import PathLike
from pathlib import Path
@ -20,7 +19,6 @@ from litellm.router_utils.batch_utils import InMemoryFile
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionAssistantMessage,
ChatCompletionCachedContent,
ChatCompletionFileObject,
ChatCompletionImageObject,
ChatCompletionReasoningItem,
@ -1807,133 +1805,6 @@ def cache_control_as_prompt_cache_breakpoint(block: _MarkedT) -> _MarkedT:
)
def prompt_cache_breakpoint_as_cache_control(block: _MarkedT) -> _MarkedT:
"""OpenAI's ``prompt_cache_breakpoint`` rewritten as Anthropic's ``cache_control`` on the same block.
A ``cache_control`` the client already wrote wins. OpenAI's 30m TTL has no Anthropic equivalent and is dropped."""
if block.get("prompt_cache_breakpoint") is None:
return block
unmarked: Final = {key: value for key, value in block.items() if key != "prompt_cache_breakpoint"}
marked: Final = (
unmarked
if unmarked.get("cache_control") is not None
else {**unmarked, "cache_control": ChatCompletionCachedContent(type="ephemeral")}
)
return cast(_MarkedT, marked) # cast-ok: same block shape as the input with the marker key renamed
def _as_mapping(value: object) -> Mapping[str, object] | None:
return value if isinstance(value, Mapping) else None # pyright: ignore[reportUnknownVariableType] # keys are read one by one
def _as_list(value: object) -> Sequence[object] | None:
return value if isinstance(value, (list, tuple)) else None # pyright: ignore[reportUnknownVariableType] # items are parsed one by one
def message_fields(message: AllMessageValues) -> Mapping[str, object]:
"""The message as a plain mapping, for code that reads keys the role-specific TypedDicts do not all declare."""
return message
def _without_key(block: Mapping[str, object], key: str) -> Mapping[str, object]:
return {name: value for name, value in block.items() if name != key}
def _block_with_prompt_cache_breakpoint_as_cache_control(block: object) -> object:
fields: Final = _as_mapping(block)
return prompt_cache_breakpoint_as_cache_control(fields) if fields is not None else block
def message_with_prompt_cache_breakpoints_as_cache_control(message: AllMessageValues) -> AllMessageValues:
"""The message-level mark, every content part's mark and every tool call's mark renamed to ``cache_control``."""
renamed: Final = prompt_cache_breakpoint_as_cache_control(message)
fields: Final = message_fields(renamed)
content: Final = _as_list(fields.get("content"))
tool_calls: Final = _as_list(fields.get("tool_calls"))
if content is None and tool_calls is None:
return renamed
return cast( # cast-ok: same message shape as the input with the marker key renamed on its blocks
AllMessageValues,
{
**fields,
**(
{"content": [_block_with_prompt_cache_breakpoint_as_cache_control(block) for block in content]}
if content is not None
else {}
),
**(
{"tool_calls": [_block_with_prompt_cache_breakpoint_as_cache_control(call) for call in tool_calls]}
if tool_calls is not None
else {}
),
},
)
def messages_with_prompt_cache_breakpoints_as_cache_control(
messages: Sequence[AllMessageValues],
) -> tuple[AllMessageValues, ...]:
return tuple(message_with_prompt_cache_breakpoints_as_cache_control(message) for message in messages)
def tool_with_prompt_cache_breakpoint_as_cache_control(tool: Mapping[str, object]) -> Mapping[str, object]:
"""A tool definition's mark, written at the top level (Anthropic shape) or under ``function`` (OpenAI shape),
renamed to a top-level ``cache_control``, which every Claude transform reads."""
function: Final = _as_mapping(tool.get("function"))
function_marked: Final = function is not None and function.get("prompt_cache_breakpoint") is not None
if tool.get("prompt_cache_breakpoint") is None and not function_marked:
return tool
unmarked_function: Final = _without_key(function, "prompt_cache_breakpoint") if function is not None else None
unmarked: Final = {
**_without_key(tool, "prompt_cache_breakpoint"),
**({"function": unmarked_function} if unmarked_function is not None else {}),
}
already_marked: Final = unmarked.get("cache_control") is not None or (
unmarked_function is not None and unmarked_function.get("cache_control") is not None
)
return unmarked if already_marked else {**unmarked, "cache_control": ChatCompletionCachedContent(type="ephemeral")}
def _block_without_cache_control(block: object) -> object:
fields: Final = _as_mapping(block)
return _without_key(fields, "cache_control") if fields is not None else block
def _blocks_without_cache_control(blocks: Sequence[object], dropped: AbstractSet[int]) -> Sequence[object]:
return [
_block_without_cache_control(block) if position in dropped else block for position, block in enumerate(blocks)
]
def message_without_cache_control_marks(
message: AllMessageValues, *, message_mark: bool, parts: AbstractSet[int], tool_calls: AbstractSet[int]
) -> AllMessageValues:
"""``message`` minus its message-level ``cache_control`` (when ``message_mark``) and the marks on the content
parts and tool calls at the given positions."""
fields: Final = message_fields(message)
kept: Final = _without_key(fields, "cache_control") if message_mark else fields
content: Final = _as_list(kept.get("content"))
calls: Final = _as_list(kept.get("tool_calls"))
return cast( # cast-ok: same message shape as the input minus the dropped marks
AllMessageValues,
{
**kept,
**({"content": _blocks_without_cache_control(content, parts)} if content is not None else {}),
**({"tool_calls": _blocks_without_cache_control(calls, tool_calls)} if calls is not None else {}),
},
)
def _tool_block_with_prompt_cache_breakpoint_as_cache_control(tool: object) -> object:
fields: Final = _as_mapping(tool)
return tool_with_prompt_cache_breakpoint_as_cache_control(fields) if fields is not None else tool
def tools_with_prompt_cache_breakpoints_as_cache_control(tools: object) -> tuple[object, ...]:
"""Every tool definition in ``tools`` (anything but a list or tuple is treated as none) with its mark renamed."""
return tuple(_tool_block_with_prompt_cache_breakpoint_as_cache_control(tool) for tool in _as_list(tools) or ())
LITELLM_INTERNAL_MESSAGE_FIELDS: Final = frozenset({"thinking_blocks", "reasoning_content", "provider_specific_fields"})

458
litellm/llms/anthropic/chat/prompt_cache_breakpoints.py
View file

@ -1,278 +1,270 @@
"""OpenAI's ``prompt_cache_breakpoint`` on a request bound for Claude.
"""OpenAI's ``prompt_cache_breakpoint`` and ``prompt_cache_options`` on a chat request bound for Claude, rewritten as
Claude's ``cache_control``."""
The marks become Anthropic's ``cache_control`` on the same blocks, and a request that carried at least one is held
to Claude's cap of four breakpoints, which a client writing the OpenAI dialect cannot see. ``prompt_cache_options``
has no Claude equivalent and is rejected or dropped like any other unsupported param."""
from collections.abc import Iterator, Mapping, Sequence
from collections.abc import Callable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from itertools import chain
from typing import Final, Literal, NoReturn, TypeAlias
from types import MappingProxyType
from typing import Final, NoReturn, TypeAlias
from typing_extensions import assert_never
from pydantic import TypeAdapter
import litellm
from litellm._logging import verbose_logger
from litellm.constants import MAX_CACHE_CONTROL_BLOCKS
from litellm.exceptions import BadRequestError, UnsupportedParamsError
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.prompt_templates.common_utils import (
message_fields,
message_without_cache_control_marks,
messages_with_prompt_cache_breakpoints_as_cache_control,
tools_with_prompt_cache_breakpoints_as_cache_control,
from litellm.exceptions import UnsupportedParamsError
from litellm.types.llms.openai import ChatCompletionCachedContent
_Path: TypeAlias = tuple[str | int, ...]
_Fields: TypeAlias = Mapping[str, object]
_Mark: TypeAlias = Callable[[_Path, _Fields], _Fields]
_MESSAGE_BLOCK_KEYS: Final = ("content", "tool_calls")
_FIELDS: Final = TypeAdapter(Mapping[str, object])
_BLOCKS: Final = TypeAdapter(Sequence[object])
_CLAUDE_TTL_MARKS: Final = MappingProxyType(
{
"5m": ChatCompletionCachedContent(type="ephemeral", ttl="5m"),
"1h": ChatCompletionCachedContent(type="ephemeral", ttl="1h"),
}
)
from litellm.llms.anthropic.common_utils import tool_call_is_rebuilt_as_server_tool_use
from litellm.types.llms.openai import AllMessageValues
PROMPT_CACHE_BREAKPOINT_ON_TOOLS: Final = "_litellm_prompt_cache_breakpoint_on_tools"
_Site: TypeAlias = tuple[Literal["tool", "message", "part", "tool_call", "tool_message"], int, int]
ClaudeWire: TypeAlias = Literal["anthropic", "bedrock_converse"]
@dataclass(frozen=True, slots=True)
class ClaudeCacheMarks:
messages: tuple[AllMessageValues, ...]
tools: tuple[object, ...]
class ClaudePromptCache:
messages: tuple[object, ...]
tools: tuple[object, ...] | None
params: _Fields
@dataclass(frozen=True, slots=True)
class CacheBreakpointCapExceeded:
found: int
class UnsupportedPromptCacheParam:
param: str
value: object
reason: str
fallback: ChatCompletionCachedContent | None
@dataclass(frozen=True, slots=True)
class PromptCacheOptionsUnsupported:
options: object
def _fields(value: object) -> _Fields | None:
return _FIELDS.validate_python(value) if isinstance(value, Mapping) else None
ClaudeCacheRefusal: TypeAlias = CacheBreakpointCapExceeded | PromptCacheOptionsUnsupported
def _blocks(value: object) -> Sequence[object] | None:
return _BLOCKS.validate_python(value) if isinstance(value, list) else None
def drop_params_enabled(litellm_params: Mapping[str, object]) -> bool:
"""``drop_params`` from the global setting, or from the request or deployment (both arrive in ``litellm_params``)."""
return bool(litellm.drop_params) or normalize_drop_params(litellm_params.get("drop_params")) is True
def _as_mapping(value: object) -> Mapping[str, object] | None:
return value if isinstance(value, Mapping) else None # pyright: ignore[reportUnknownVariableType] # keys are read one by one
def _as_blocks(value: object) -> Sequence[object] | None:
return value if isinstance(value, (list, tuple)) else None # pyright: ignore[reportUnknownVariableType] # blocks are parsed one by one
def _carries(block: object, key: str) -> bool:
fields: Final = _as_mapping(block)
return fields is not None and fields.get(key) is not None
def _tool_carries(tool: object, key: str) -> bool:
fields: Final = _as_mapping(tool)
return fields is not None and (fields.get(key) is not None or _carries(fields.get("function"), key))
def tools_carry_prompt_cache_breakpoint(tools: object) -> bool:
return any(_tool_carries(tool, "prompt_cache_breakpoint") for tool in _as_blocks(tools) or ())
def _renders_message_mark(message: Mapping[str, object], wire: ClaudeWire) -> bool:
"""Claude's message builder keeps a message-level mark on string content and on tool results only. Anthropic's
builder skips an empty assistant string, mark included, where Converse still emits its ``cachePoint``."""
content: Final = message.get("content")
if isinstance(content, str):
return bool(content) or message.get("role") != "assistant" or wire == "bedrock_converse"
return message.get("role") in ("tool", "function")
def _renders_tool_call_mark(message: Mapping[str, object], call: object, wire: ClaudeWire) -> bool:
"""Anthropic rebuilds a tool call answered by a server tool as ``server_tool_use``, which carries no mark."""
return wire == "bedrock_converse" or not tool_call_is_rebuilt_as_server_tool_use(
(_as_mapping(call) or {}).get("id"), message.get("provider_specific_fields")
)
def _message_sites(index: int, message: object, key: str, wire: ClaudeWire) -> Iterator[_Site]:
fields: Final = _as_mapping(message)
if fields is None:
return
if fields.get(key) is not None and _renders_message_mark(fields, wire):
yield ("message", index, -1)
for position, part in enumerate(_as_blocks(fields.get("content")) or ()):
if _carries(part, key):
yield ("part", index, position)
for position, call in enumerate(_as_blocks(fields.get("tool_calls")) or ()):
if _carries(call, key) and _renders_tool_call_mark(fields, call, wire):
yield ("tool_call", index, position)
def _converse_message_sites(index: int, message: object, key: str) -> Iterator[_Site]:
"""Converse emits a single ``cachePoint`` per tool message, however many of its blocks carry a mark."""
fields: Final = _as_mapping(message)
if fields is None or fields.get("role") != "tool":
yield from _message_sites(index, message, key, "bedrock_converse")
return
if fields.get(key) is not None or any(_carries(part, key) for part in _as_blocks(fields.get("content")) or ()):
yield ("tool_message", index, -1)
def _sites(messages: Sequence[object], tools: Sequence[object], key: str, wire: ClaudeWire) -> tuple[_Site, ...]:
"""Every mark of ``key`` the ``wire`` renders, in wire order: tool definitions first, then the messages as
listed. Converse reads only a tool definition's top-level mark."""
if wire == "bedrock_converse":
return (
*(("tool", index, -1) for index, tool in enumerate(tools) if _carries(tool, key)),
*chain.from_iterable(
_converse_message_sites(index, message, key) for index, message in enumerate(messages)
),
)
def _ttl_mark(ttl: object, param: str, value: object) -> ChatCompletionCachedContent | UnsupportedPromptCacheParam:
if ttl is None:
return ChatCompletionCachedContent(type="ephemeral")
known: Final = _CLAUDE_TTL_MARKS.get(ttl) if isinstance(ttl, str) else None
return (
*(("tool", index, -1) for index, tool in enumerate(tools) if _tool_carries(tool, key)),
*chain.from_iterable(_message_sites(index, message, key, wire) for index, message in enumerate(messages)),
ChatCompletionCachedContent(**known)
if known is not None
else UnsupportedPromptCacheParam(
param,
value,
"Claude caches for a ttl of 5m or 1h only",
fallback=ChatCompletionCachedContent(type="ephemeral"),
)
)
def _without_cache_control(block: Mapping[str, object]) -> Mapping[str, object]:
return {key: value for key, value in block.items() if key != "cache_control"}
def _breakpoint_mark(marker: object) -> ChatCompletionCachedContent | UnsupportedPromptCacheParam:
fields: Final = _fields(marker)
if fields is None or fields.get("mode") != "explicit":
return UnsupportedPromptCacheParam(
"prompt_cache_breakpoint", marker, 'expected {"mode": "explicit"} with an optional ttl', fallback=None
)
return _ttl_mark(fields.get("ttl"), "prompt_cache_breakpoint", marker)
def _tool_without_cache_control(tool: object) -> object:
fields: Final = _as_mapping(tool)
def _automatic_mark(
options: object, accepts_automatic_caching: bool
) -> ChatCompletionCachedContent | UnsupportedPromptCacheParam | None:
if options is None:
return None
fields: Final = _fields(options)
if fields is None or fields.get("mode") not in (None, "implicit", "explicit"):
return UnsupportedPromptCacheParam(
"prompt_cache_options", options, 'expected a "mode" of "implicit" or "explicit"', fallback=None
)
if fields.get("mode") != "implicit":
return (
None
if fields.get("ttl") is None
else UnsupportedPromptCacheParam(
"prompt_cache_options", options, 'Claude applies a ttl only with "mode": "implicit"', fallback=None
)
)
if not accepts_automatic_caching:
return UnsupportedPromptCacheParam(
"prompt_cache_options", options, "this route has no automatic caching", fallback=None
)
return _ttl_mark(fields.get("ttl"), "prompt_cache_options", options)
def _indexed_fields(values: Sequence[object]) -> Iterator[tuple[int, _Fields]]:
return ((index, fields) for index, fields in enumerate(map(_fields, values)) if fields is not None)
def _markable_in_message(index: int, message: _Fields) -> Iterator[tuple[_Path, _Fields]]:
yield ("messages", index), message
for key in _MESSAGE_BLOCK_KEYS:
yield from (
(("messages", index, key, j), block) for j, block in _indexed_fields(_blocks(message.get(key)) or ())
)
def _markable_in_tool(index: int, tool: _Fields) -> Iterator[tuple[_Path, _Fields]]:
yield ("tools", index), tool
function: Final = _fields(tool.get("function"))
if function is not None:
yield ("tools", index, "function"), function
def _markable(messages: Sequence[object], tools: Sequence[object]) -> Iterator[tuple[_Path, _Fields]]:
"""Every block that can carry a cache mark, in the order Claude renders them: tool definitions, then messages."""
for index, tool in _indexed_fields(tools):
yield from _markable_in_tool(index, tool)
for index, message in _indexed_fields(messages):
yield from _markable_in_message(index, message)
def _marked_block(path: _Path, block: object, mark: _Mark) -> object:
fields: Final = _fields(block)
return block if fields is None else mark(path, fields)
def _marked_blocks(message: _Fields, path: _Path, key: str, mark: _Mark) -> _Fields:
blocks: Final = _blocks(message.get(key))
if blocks is None:
return {}
return {key: [_marked_block((*path, key, j), block, mark) for j, block in enumerate(blocks)]}
def _marked_message(index: int, message: object, mark: _Mark) -> object:
fields: Final = _fields(message)
if fields is None:
return tool
function: Final = _as_mapping(fields.get("function"))
return message
marked: Final = mark(("messages", index), fields)
return {
**_without_cache_control(fields),
**({"function": _without_cache_control(function)} if function is not None else {}),
**marked,
**_marked_blocks(marked, ("messages", index), "content", mark),
**_marked_blocks(marked, ("messages", index), "tool_calls", mark),
}
def _dropped_positions(dropped: frozenset[_Site], kind: str, index: int) -> frozenset[int]:
return frozenset(
position for site_kind, site_index, position in dropped if (site_kind, site_index) == (kind, index)
)
def _marked_tool(index: int, tool: object, mark: _Mark) -> object:
fields: Final = _fields(tool)
if fields is None:
return tool
marked: Final = mark(("tools", index), fields)
function: Final = _fields(marked.get("function"))
return marked if function is None else {**marked, "function": mark(("tools", index, "function"), function)}
def _message_without(index: int, message: AllMessageValues, dropped: frozenset[_Site]) -> AllMessageValues:
if ("tool_message", index, -1) in dropped:
content: Final = _as_blocks(message_fields(message).get("content")) or ()
return message_without_cache_control_marks(
message, message_mark=True, parts=frozenset(range(len(content))), tool_calls=frozenset()
)
return message_without_cache_control_marks(
message,
message_mark=("message", index, -1) in dropped,
parts=_dropped_positions(dropped, "part", index),
tool_calls=_dropped_positions(dropped, "tool_call", index),
)
def _prompt_cache_options_refusal(options: object, drop_params: bool) -> PromptCacheOptionsUnsupported | None:
if options is None:
def _breakpoint_problem(block: _Fields) -> UnsupportedPromptCacheParam | None:
marker: Final = block.get("prompt_cache_breakpoint")
if marker is None or block.get("cache_control") is not None:
return None
if not drop_params:
return PromptCacheOptionsUnsupported(options)
verbose_logger.debug("Dropping prompt_cache_options=%r: Claude caches through cache_control.", options)
return None
mark: Final = _breakpoint_mark(marker)
return mark if isinstance(mark, UnsupportedPromptCacheParam) else None
def claude_cache_marks(
messages: Sequence[AllMessageValues],
tools: object,
def _renamed(_path: _Path, block: _Fields) -> _Fields:
"""A ``cache_control`` the client already wrote on the block wins over the breakpoint."""
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"}
if unmarked.get("cache_control") is not None:
return unmarked
mark: Final = _breakpoint_mark(marker)
cache_control: Final = mark.fallback if isinstance(mark, UnsupportedPromptCacheParam) else mark
return unmarked if cache_control is None else {**unmarked, "cache_control": cache_control}
def _marked_request(
messages: Sequence[object], tools: Sequence[object], mark: _Mark
) -> tuple[tuple[object, ...], tuple[object, ...]]:
return (
tuple(_marked_message(index, message, mark) for index, message in enumerate(messages)),
tuple(_marked_tool(index, tool, mark) for index, tool in enumerate(tools)),
)
def _without_cache_control(dropped: frozenset[_Path]) -> _Mark:
def mark(path: _Path, block: _Fields) -> _Fields:
return {key: value for key, value in block.items() if key != "cache_control"} if path in dropped else block
return mark
def claude_prompt_cache(
messages: Sequence[object],
tools: Sequence[object] | None,
params: _Fields,
*,
automatic_breakpoint: bool,
accepts_automatic_caching: bool,
drop_params: bool,
prompt_cache_options: object = None,
tools_carried_breakpoint: bool = False,
wire: ClaudeWire = "anthropic",
) -> ClaudeCacheMarks | ClaudeCacheRefusal:
"""The request's ``prompt_cache_breakpoint`` marks renamed to ``cache_control`` and, when it carried any, held to
Claude's cap: ``drop_params`` keeps the latest four in wire order, otherwise the excess is reported. A request
that carried none is forwarded as written so the provider's own answer reaches the client.
``prompt_cache_options`` has no Claude equivalent: the caller has already taken it out of the request, and it is
reported unless ``drop_params``. ``automatic_breakpoint`` is the request-level ``cache_control`` (Anthropic's
automatic caching), which takes one slot of its own and is never dropped. ``tools_carried_breakpoint`` covers
tools already mapped to Anthropic's shape, where the mark was renamed before this census ran. ``wire`` is the
body the marks are rendered into, so the count matches the breakpoints the provider receives."""
options_refusal: Final = _prompt_cache_options_refusal(prompt_cache_options, drop_params)
if options_refusal is not None:
return options_refusal
listed_tools: Final = _as_blocks(tools) or ()
if not tools_carried_breakpoint and not _sites(messages, listed_tools, "prompt_cache_breakpoint", wire):
return ClaudeCacheMarks(tuple(messages), tuple(listed_tools))
renamed_messages: Final = messages_with_prompt_cache_breakpoints_as_cache_control(messages)
renamed_tools: Final = tools_with_prompt_cache_breakpoints_as_cache_control(listed_tools)
sites: Final = _sites(renamed_messages, renamed_tools, "cache_control", wire)
found: Final = len(sites) + int(automatic_breakpoint)
if found <= MAX_CACHE_CONTROL_BLOCKS:
return ClaudeCacheMarks(renamed_messages, renamed_tools)
if not drop_params:
return CacheBreakpointCapExceeded(found)
dropped: Final = frozenset(sites[: found - MAX_CACHE_CONTROL_BLOCKS])
verbose_logger.debug(
"Dropping the earliest %d of %d cache breakpoints: Claude accepts at most %d.",
len(dropped),
found,
MAX_CACHE_CONTROL_BLOCKS,
) -> ClaudePromptCache | UnsupportedPromptCacheParam:
"""``prompt_cache_breakpoint`` marks renamed to ``cache_control``, and ``prompt_cache_options`` with
``"mode": "implicit"`` turned into the request-level ``cache_control`` (Claude's automatic caching). A mark or
option Claude cannot honor is reported unless ``drop_params``, which keeps what Claude can use and, when the
request ends up with more than Claude's four breakpoints, the latest four. A request carrying neither param is
returned as written."""
listed_tools: Final = tools or ()
options: Final = params.get("prompt_cache_options")
blocks: Final = tuple(block for _, block in _markable(messages, listed_tools))
if options is None and all(block.get("prompt_cache_breakpoint") is None for block in blocks):
return ClaudePromptCache(tuple(messages), None if tools is None else tuple(tools), params)
automatic: Final = _automatic_mark(options, accepts_automatic_caching)
problem: Final = (
automatic
if isinstance(automatic, UnsupportedPromptCacheParam)
else next(filter(None, map(_breakpoint_problem, blocks)), None)
)
return ClaudeCacheMarks(
messages=tuple(_message_without(index, message, dropped) for index, message in enumerate(renamed_messages)),
tools=tuple(
_tool_without_cache_control(tool) if ("tool", index, -1) in dropped else tool
for index, tool in enumerate(renamed_tools)
if problem is not None and not drop_params:
return problem
if problem is not None:
verbose_logger.debug("Claude: adjusting %s=%r because %s", problem.param, problem.value, problem.reason)
automatic_mark: Final = automatic.fallback if isinstance(automatic, UnsupportedPromptCacheParam) else automatic
request_params: Final = {
**{key: value for key, value in params.items() if key != "prompt_cache_options"},
**(
{"cache_control": automatic_mark}
if automatic_mark is not None and params.get("cache_control") is None
else {}
),
}
renamed_messages, renamed_tools = _marked_request(messages, listed_tools, _renamed)
sites: Final = tuple(
path for path, block in _markable(renamed_messages, renamed_tools) if block.get("cache_control") is not None
)
excess: Final = len(sites) + int(request_params.get("cache_control") is not None) - MAX_CACHE_CONTROL_BLOCKS
if excess > 0 and drop_params:
verbose_logger.debug("Claude: dropping the earliest %d cache breakpoints", excess)
trimmed_messages, trimmed_tools = (
_marked_request(renamed_messages, renamed_tools, _without_cache_control(frozenset(sites[:excess])))
if excess > 0 and drop_params
else (renamed_messages, renamed_tools)
)
return ClaudePromptCache(trimmed_messages, None if tools is None else trimmed_tools, request_params)
def claude_extra_body(
extra_body: Mapping[str, object], drop_params: bool
) -> Mapping[str, object] | PromptCacheOptionsUnsupported:
"""``extra_body`` merged into a Claude body, held to the same ``prompt_cache_options`` rule as the request."""
refusal: Final = _prompt_cache_options_refusal(extra_body.get("prompt_cache_options"), drop_params)
if refusal is not None:
return refusal
if "prompt_cache_options" not in extra_body:
return extra_body
return {key: value for key, value in extra_body.items() if key != "prompt_cache_options"}
def raise_claude_cache_refusal(refusal: ClaudeCacheRefusal, model: str, llm_provider: str) -> NoReturn:
match refusal:
case CacheBreakpointCapExceeded(found=found):
raise BadRequestError(
message=(
f"Claude accepts at most {MAX_CACHE_CONTROL_BLOCKS} cache breakpoints, found {found}. "
"Set drop_params=True (proxy: litellm_settings: drop_params: true) to keep the latest four."
),
model=model,
llm_provider=llm_provider,
)
case PromptCacheOptionsUnsupported(options=options):
raise UnsupportedParamsError(
message=(
f"{model} does not support prompt_cache_options={options!r}. It caches through cache_control "
"marks (or prompt_cache_breakpoint on the OpenAI dialect). To drop unsupported params, set "
"`litellm.drop_params = True`."
),
model=model,
status_code=400,
)
case _:
assert_never(refusal)
def resolved_claude_cache_marks(
outcome: ClaudeCacheMarks | ClaudeCacheRefusal, model: str, llm_provider: str
) -> ClaudeCacheMarks:
if isinstance(outcome, ClaudeCacheMarks):
def resolved_claude_prompt_cache(
outcome: ClaudePromptCache | UnsupportedPromptCacheParam, model: str, llm_provider: str
) -> ClaudePromptCache:
if isinstance(outcome, ClaudePromptCache):
return outcome
raise_claude_cache_refusal(outcome, model=model, llm_provider=llm_provider)
raise_unsupported_prompt_cache_param(outcome, model=model, llm_provider=llm_provider)
def resolved_claude_extra_body(
outcome: Mapping[str, object] | PromptCacheOptionsUnsupported, model: str, llm_provider: str
) -> Mapping[str, object]:
if isinstance(outcome, PromptCacheOptionsUnsupported):
raise_claude_cache_refusal(outcome, model=model, llm_provider=llm_provider)
return outcome
def raise_unsupported_prompt_cache_param(
problem: UnsupportedPromptCacheParam, model: str, llm_provider: str
) -> NoReturn:
raise UnsupportedParamsError(
message=(
f"{model} does not support {problem.param}={problem.value!r}: {problem.reason}. "
"To drop unsupported params, set `litellm.drop_params = True`."
),
model=model,
llm_provider=llm_provider,
)

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

@ -25,7 +25,6 @@ from litellm.constants import (
from litellm.litellm_core_utils.core_helpers import map_finish_reason
from litellm.litellm_core_utils.prompt_templates.common_utils import (
sanitize_input_schema_for_anthropic,
tools_with_prompt_cache_breakpoints_as_cache_control,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
RemoteMedia,
@ -111,15 +110,6 @@ from ..common_utils import (
requires_native_compaction_beta,
strip_advisor_blocks_from_messages,
)
from .prompt_cache_breakpoints import (
PROMPT_CACHE_BREAKPOINT_ON_TOOLS,
claude_cache_marks,
claude_extra_body,
drop_params_enabled,
resolved_claude_cache_marks,
resolved_claude_extra_body,
tools_carry_prompt_cache_breakpoint,
)
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -1578,19 +1568,12 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if param == "max_tokens" or param == "max_completion_tokens":
optional_params["max_tokens"] = value if isinstance(value, int) else max(1, int(round(value)))
elif param == "tools":
requested_tools = cast(object, value) # cast-ok: the loop yields untyped values
anthropic_tools, mcp_servers = self._map_tools(
list(tools_with_prompt_cache_breakpoints_as_cache_control(requested_tools))
if tools_carry_prompt_cache_breakpoint(requested_tools)
else value
)
anthropic_tools, mcp_servers = self._map_tools(value)
optional_params = self._add_tools_to_optional_params(
optional_params=optional_params, tools=anthropic_tools
)
if mcp_servers:
optional_params["mcp_servers"] = mcp_servers
if tools_carry_prompt_cache_breakpoint(requested_tools):
optional_params[PROMPT_CACHE_BREAKPOINT_ON_TOOLS] = True # rebind-ok: out-param store
elif param == "tool_choice" or param == "parallel_tool_calls":
_tool_choice: AnthropicMessagesToolChoice | None = self._map_tool_choice(
tool_choice=non_default_params.get("tool_choice"),
@ -1987,19 +1970,6 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
headers=headers,
)
def transform_extra_body(
self,
extra_body: Mapping[str, object],
request: Mapping[str, object],
model: str,
litellm_params: Mapping[str, object],
) -> Mapping[str, object]:
return resolved_claude_extra_body(
claude_extra_body(extra_body, drop_params=drop_params_enabled(litellm_params)),
model=model,
llm_provider=self._resolved_provider,
)
def transform_request(
self,
model: str,
@ -2053,22 +2023,6 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
custom_llm_provider=self._resolved_provider,
)
body: Final = cast(dict[str, object], optional_params) # cast-ok: BaseConfig passes an untyped dict
cache_marks: Final = resolved_claude_cache_marks(
claude_cache_marks(
messages=messages,
tools=body.get("tools"),
automatic_breakpoint=body.get("cache_control") is not None,
drop_params=drop_params_enabled(litellm_params),
prompt_cache_options=body.pop("prompt_cache_options", None),
tools_carried_breakpoint=bool(body.pop(PROMPT_CACHE_BREAKPOINT_ON_TOOLS, False)),
),
model=model,
llm_provider=self._resolved_provider,
)
if "tools" in body:
optional_params["tools"] = list(cache_marks.tools) # rebind-ok: out-param store
# === Tool-name sanitization (single chokepoint) ===
# Anthropic enforces ^[a-zA-Z0-9_-]{1,128}$ on every tool name. We
# sanitize *here* -- not in map_openai_params -- because:
@ -2092,18 +2046,15 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
_name_forward_map, _name_reverse_map = self._sanitize_tool_names_in_request(
optional_params=optional_params,
)
marked_messages: Final = (
self._rewrite_tool_names_in_messages(list(cache_marks.messages), _name_forward_map)
if _name_forward_map
else list(cache_marks.messages)
)
if _name_forward_map:
messages = self._rewrite_tool_names_in_messages(messages, _name_forward_map)
if _name_reverse_map and isinstance(litellm_params, dict):
litellm_params[ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY] = _name_reverse_map
# Only the leading system run becomes the top-level system prompt. A later
# system message stays in the conversation: hoisting it rewrites the cached
# prefix and re-bills the whole history at cache-write pricing (#36559).
leading_system_run, later_messages = split_leading_system_run(marked_messages)
leading_system_run, later_messages = split_leading_system_run(messages)
anthropic_system_message_list: Final = self.translate_system_message(messages=list(leading_system_run))
# Handling anthropic API Prompt Caching
if len(anthropic_system_message_list) > 0:
@ -2124,7 +2075,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
except Exception as e:
raise AnthropicError(
status_code=400,
message=f"{e}\nReceived Messages={marked_messages}",
message=f"{e}\nReceived Messages={messages}",
) # don't use verbose_logger.exception, if exception is raised
self.update_headers_with_optional_anthropic_beta(

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

@ -44,11 +44,6 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
message_field,
parts_of,
)
from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
claude_cache_marks,
drop_params_enabled,
resolved_claude_cache_marks,
)
from litellm.llms.anthropic.chat.transformation import (
DROP_UNSUPPORTED_ADAPTIVE_THINKING_WARNING,
DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
@ -2017,34 +2012,6 @@ class AmazonConverseConfig(BaseConfig):
return data
def _messages_with_prompt_cache_breakpoints(
self,
model: str,
messages: Sequence[AllMessageValues],
optional_params: dict[str, object], # mutable-ok: the request body is built from this out-param
litellm_params: Mapping[str, object],
) -> tuple[AllMessageValues, ...]:
"""On a model that takes ``cachePoint`` blocks, OpenAI ``prompt_cache_breakpoint`` marks as ``cache_control``
so the factory emits them, held to the cap of four, with ``prompt_cache_options`` rejected or dropped. Any
other model gets the request as written."""
if not bedrock_model_accepts_cache_points(model):
return tuple(messages)
cache_marks: Final = resolved_claude_cache_marks(
claude_cache_marks(
messages=messages,
tools=optional_params.get("tools"),
automatic_breakpoint=optional_params.get("cache_control") is not None,
drop_params=drop_params_enabled(litellm_params),
prompt_cache_options=optional_params.pop("prompt_cache_options", None),
wire="bedrock_converse",
),
model=model,
llm_provider=self.custom_llm_provider or "bedrock",
)
if "tools" in optional_params:
optional_params["tools"] = list(cache_marks.tools) # rebind-ok: out-param store
return cache_marks.messages
async def _async_transform_request(
self,
model: str,
@ -2053,11 +2020,7 @@ class AmazonConverseConfig(BaseConfig):
litellm_params: dict,
headers: dict | None = None,
) -> RequestObject:
body: Final = cast(dict[str, object], optional_params) # cast-ok: BaseConfig passes an untyped dict
params: Final = cast(Mapping[str, object], litellm_params) # cast-ok: BaseConfig passes an untyped dict
messages, system_content_blocks = self._transform_system_message(
list(self._messages_with_prompt_cache_breakpoints(model, messages, body, params)), model=model
)
messages, system_content_blocks = self._transform_system_message(messages, model=model)
# Convert last user message to guarded_text if guardrailConfig is present
messages = self._convert_consecutive_user_messages_to_guarded_text(
@ -2125,11 +2088,7 @@ class AmazonConverseConfig(BaseConfig):
litellm_params: dict,
headers: dict | None = None,
) -> RequestObject:
body: Final = cast(dict[str, object], optional_params) # cast-ok: BaseConfig passes an untyped dict
params: Final = cast(Mapping[str, object], litellm_params) # cast-ok: BaseConfig passes an untyped dict
messages, system_content_blocks = self._transform_system_message(
list(self._messages_with_prompt_cache_breakpoints(model, messages, body, params)), model=model
)
messages, system_content_blocks = self._transform_system_message(messages, model=model)
# Convert last user message to guarded_text if guardrailConfig is present
messages = self._convert_consecutive_user_messages_to_guarded_text(

75
litellm/llms/databricks/chat/transformation.py
View file

@ -55,15 +55,6 @@ from litellm.types.utils import (
Usage,
)
from ...anthropic.chat.prompt_cache_breakpoints import (
PROMPT_CACHE_BREAKPOINT_ON_TOOLS,
claude_cache_marks,
claude_extra_body,
drop_params_enabled,
resolved_claude_cache_marks,
resolved_claude_extra_body,
tools_carry_prompt_cache_breakpoint,
)
from ...anthropic.chat.transformation import (
REASONING_EFFORT_TO_OUTPUT_CONFIG_EFFORT,
AnthropicConfig,
@ -384,9 +375,6 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
)
if "tools" in mapped_params:
mapped_params["tools"] = self._map_openai_to_dbrx_tool(model=model, tools=mapped_params["tools"])
requested_params: Final = cast(Mapping[str, object], non_default_params) # cast-ok: base signature is untyped
if "claude" in model and tools_carry_prompt_cache_breakpoint(requested_params.get("tools")):
mapped_params[PROMPT_CACHE_BREAKPOINT_ON_TOOLS] = True
if "max_completion_tokens" in non_default_params and replace_max_completion_tokens_with_max_tokens:
mapped_params["max_tokens"] = non_default_params[
"max_completion_tokens"
@ -446,69 +434,6 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
return mapped_params
def transform_request(
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: overrides BaseConfig.transform_request
optional_params: dict[str, object], # mutable-ok: overrides BaseConfig.transform_request
litellm_params: dict[str, object], # mutable-ok: overrides BaseConfig.transform_request
headers: dict[str, object], # mutable-ok: overrides BaseConfig.transform_request
) -> dict[str, object]: # mutable-ok: overrides BaseConfig.transform_request
tools_carried_breakpoint: Final = bool(optional_params.pop(PROMPT_CACHE_BREAKPOINT_ON_TOOLS, False))
marked_messages: Final = (
self._messages_with_claude_cache_marks(
model, messages, optional_params, litellm_params, tools_carried_breakpoint
)
if "claude" in model
else messages
)
return cast( # cast-ok: OpenAIGPTConfig.transform_request returns an untyped dict
dict[str, object],
super().transform_request( # pyright: ignore[reportUnknownMemberType] # base signature uses bare dict
model, list(marked_messages), optional_params, litellm_params, headers
),
)
def transform_extra_body(
self,
extra_body: Mapping[str, object],
request: Mapping[str, object],
model: str,
litellm_params: Mapping[str, object],
) -> Mapping[str, object]:
if "claude" not in model:
return extra_body
return resolved_claude_extra_body(
claude_extra_body(extra_body, drop_params=drop_params_enabled(litellm_params)),
model=model,
llm_provider="databricks",
)
def _messages_with_claude_cache_marks(
self,
model: str,
messages: Sequence[AllMessageValues],
optional_params: dict[str, object], # mutable-ok: the request body is built from this out-param
litellm_params: Mapping[str, object],
tools_carried_breakpoint: bool,
) -> tuple[AllMessageValues, ...]:
"""OpenAI ``prompt_cache_breakpoint`` marks as the ``cache_control`` Databricks forwards to Claude, held to
Claude's cap, with ``prompt_cache_options`` rejected or dropped. Tool definitions reach Databricks without
any mark, so only the message marks count toward the cap."""
cache_marks: Final = resolved_claude_cache_marks(
claude_cache_marks(
messages=messages,
tools=(),
automatic_breakpoint=optional_params.get("cache_control") is not None,
drop_params=drop_params_enabled(litellm_params),
prompt_cache_options=optional_params.pop("prompt_cache_options", None),
tools_carried_breakpoint=tools_carried_breakpoint,
),
model=model,
llm_provider="databricks",
)
return cache_marks.messages
def _should_fake_stream(self, optional_params: dict) -> bool:
"""
Databricks doesn't support 'response_format' while streaming

37
litellm/main.py
View file

@ -210,6 +210,8 @@ from .litellm_core_utils.streaming_chunk_builder_utils import (
ChunkProcessor as ChunkProcessor, # noqa: PLC0414 # public re-export
)
from .llms.anthropic.chat import AnthropicChatCompletion
from .llms.anthropic.chat.prompt_cache_breakpoints import claude_prompt_cache, resolved_claude_prompt_cache
from .llms.anthropic.chat.transformation import AnthropicConfig
from .llms.azure.audio_transcriptions import AzureAudioTranscription
from .llms.azure.azure import AzureChatCompletion, check_dynamic_azure_params
from .llms.azure.chat.o_series_handler import AzureOpenAIO1ChatCompletion
@ -5555,8 +5557,35 @@ def completion(
base_model=base_model,
)
claude_drop_params: Final = litellm.drop_params is True or (
"drop_params" in kwargs
and normalize_drop_params(cast(object, kwargs["drop_params"])) is True # cast-ok: untyped request kwargs
)
claude_cache: Final = (
resolved_claude_prompt_cache(
claude_prompt_cache(
cast(list[object], messages), # cast-ok: completion types messages as a bare List
cast(list[object] | None, tools), # cast-ok: completion types tools as a bare List
non_default_params,
accepts_automatic_caching=isinstance(provider_config, AnthropicConfig),
drop_params=claude_drop_params,
),
model=model,
llm_provider=custom_llm_provider,
)
if provider_config is not None
and _is_claude_tool_target(custom_llm_provider=custom_llm_provider, model=model)
else None
)
claude_messages: Final = (
None
if claude_cache is None
else cast(list[AllMessageValues], list(claude_cache.messages)) # cast-ok: marks renamed in place
)
if provider_config is not None:
messages = provider_config.translate_developer_role_to_system_role(messages=messages)
messages = provider_config.translate_developer_role_to_system_role(
messages=messages if claude_messages is None else claude_messages
)
if (
supports_system_message is not None
@ -5599,7 +5628,7 @@ def completion(
"custom_llm_provider": custom_llm_provider,
"response_format": response_format,
"seed": seed,
"tools": tools,
"tools": tools if claude_cache is None or claude_cache.tools is None else list(claude_cache.tools),
"tool_choice": tool_choice,
"max_retries": max_retries,
"logprobs": logprobs,
@ -5623,7 +5652,9 @@ def completion(
"allowed_openai_params": allowed_openai_params,
"base_model": base_model,
}
optional_params = get_optional_params(**optional_param_args, **non_default_params)
optional_params = get_optional_params(
**optional_param_args, **(non_default_params if claude_cache is None else claude_cache.params)
)
processed_non_default_params: Final = pre_process_non_default_params(
model=model,
passed_params=optional_param_args,

43
tests/e2e/llm_translation/test_cache_control.py
View file

@ -36,8 +36,9 @@ Each case asserts the feature actually happened, not just a 200. Coverage matrix
system message, a user part, a tool result, a tool call or a tool definition,
or beside a ``cache_control`` on the same system part, must write the cache
on the first call and read it on the second. On
anthropic and Converse, a 30m breakpoint, five breakpoints and
``prompt_cache_options`` under drop_params must still cache.
anthropic and Converse, a 30m breakpoint and five breakpoints under
drop_params must still cache. On anthropic, ``prompt_cache_options`` with
``"mode": "implicit"`` must cache with no breakpoint at all.
service_tier lives in test_provider_features_e2e.py.
@ -110,9 +111,7 @@ def _cacheable_prefix() -> str:
"""A prefix long enough to clear provider minimum cacheable sizes (Haiku is
2048 tokens), unique per run so the first call writes and the second reads."""
marker = unique_marker()
body = " ".join(
f"Cacheable reference paragraph {index} for run {marker}." for index in range(600)
)
body = " ".join(f"Cacheable reference paragraph {index} for run {marker}." for index in range(600))
return f"{body}\nEnd of reference material {marker}."
@ -168,9 +167,7 @@ def _plain_cache_chat(
)
def _assert_cache_read_on_second_call(
model: str, send: Callable[[str], Result[ChatResponse]]
) -> None:
def _assert_cache_read_on_second_call(model: str, send: Callable[[str], Result[ChatResponse]]) -> None:
prefix = _cacheable_prefix()
first = unwrap(send(prefix))
@ -296,9 +293,7 @@ class TestCacheControl:
mode=Mode.NONSTREAM,
)
)
def test_bedrock_prompt_caching_reads_cache(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
def test_bedrock_prompt_caching_reads_cache(self, client: PassthroughClient, resources: ResourceManager) -> None:
model = f"e2e-bedrock-cache-{unique_marker()}"
model_id = client.proxy.create_model(
model,
@ -322,9 +317,7 @@ class TestCacheControl:
mode=Mode.NONSTREAM,
)
)
def test_vertex_prompt_caching_reads_cache(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
def test_vertex_prompt_caching_reads_cache(self, client: PassthroughClient, resources: ResourceManager) -> None:
model = f"e2e-vertex-cache-{unique_marker()}"
model_id = client.proxy.create_model(
model,
@ -356,9 +349,7 @@ class TestCacheControl:
mode=Mode.NONSTREAM,
)
)
def test_anthropic_prompt_caching_reads_cache(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
def test_anthropic_prompt_caching_reads_cache(self, client: PassthroughClient, resources: ResourceManager) -> None:
model = f"e2e-anthropic-cache-{unique_marker()}"
model_id = client.proxy.create_model(
model,
@ -455,9 +446,7 @@ class TestCacheControl:
mode=Mode.NONSTREAM,
)
)
def test_openai_prompt_caching_reads_cache(
self, client: PassthroughClient, resources: ResourceManager
) -> None:
def test_openai_prompt_caching_reads_cache(self, client: PassthroughClient, resources: ResourceManager) -> None:
model = f"e2e-openai-cache-{unique_marker()}"
model_id = client.proxy.create_model(
model,
@ -709,8 +698,8 @@ def _assert_claude_writes_then_reads_cache(client: PassthroughClient, key: str,
class TestPromptCacheBreakpointOnClaude:
"""OpenAI's ``prompt_cache_breakpoint`` sent to a Claude deployment caches the prefix it marks.
The calls that leave drop_params off (more than four breakpoints, or ``prompt_cache_options``) answer 400 before
any provider call, so the wire tests in tests/integration prove them: the shared stack sets
The 400s that need drop_params off (a 30m ttl, or implicit ``prompt_cache_options`` on Converse) happen before any
provider call, so the wire tests in tests/integration prove them: the shared stack sets
``litellm_settings.drop_params: true``, which no deployment can switch back off."""
@pytest.mark.parametrize(
@ -766,17 +755,17 @@ class TestPromptCacheBreakpointOnClaude:
)
_assert_claude_writes_then_reads_cache(client, scoped_key, body)
@pytest.mark.parametrize("route", [_claude_route_param(route) for route in ("anthropic", "bedrock_converse")])
def test_prompt_cache_options_are_dropped_under_drop_params(
@pytest.mark.parametrize("route", [_claude_route_param("anthropic")])
def test_implicit_prompt_cache_options_cache_without_any_breakpoint(
self, client: PassthroughClient, resources: ResourceManager, scoped_key: str, route: ClaudeRoute
) -> None:
model: Final = _register_claude(client, resources, route, drop_params=True)
model: Final = _register_claude(client, resources, route)
body: Final = ChatBody(
model=model,
max_tokens=64,
prompt_cache_options=PromptCacheOptions(mode="explicit"),
prompt_cache_options=PromptCacheOptions(mode="implicit"),
messages=[
ChatMessage(role="system", content=[_marked(_cacheable_prefix())]),
ChatMessage(role="system", content=_cacheable_prefix()),
ChatMessage(role="user", content="Reply with one word."),
],
)

270
tests/integration/providers/test_claude_chat_prompt_cache_breakpoint_wire.py
View file

@ -1,7 +1,8 @@
import json
from collections.abc import Iterator, Mapping
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal, assert_never
from typing import Final, Literal
import pytest
from integration._support.client import Gateway, Scenario, gateway_from_environment
@ -37,12 +38,13 @@ from integration.providers._cache_control_marks_support import (
tool_call,
tool_use_label,
)
from pydantic import BaseModel, ConfigDict, JsonValue
from pydantic import JsonValue
from typing_extensions import assert_never
Provider = Literal[
"anthropic", "bedrock_converse", "bedrock_invoke", "bedrock_claude_platform", "vertex_ai", "azure_ai"
]
Site = Literal["tool", "system", "user_part", "tool_call", "tool_result_part", "string_message"]
Site = Literal["tool", "system", "user_part", "tool_call", "tool_result", "string_message"]
_PROVIDERS: Final[tuple[Provider, ...]] = (
"anthropic",
@ -52,84 +54,26 @@ _PROVIDERS: Final[tuple[Provider, ...]] = (
"vertex_ai",
"azure_ai",
)
_EXTRA_BODY_PROVIDERS: Final[tuple[Provider, ...]] = ("bedrock_invoke", "bedrock_claude_platform", "azure_ai")
_SITES: Final[tuple[Site, ...]] = ("tool", "system", "user_part", "tool_call", "tool_result_part", "string_message")
_AUTOMATIC_CACHING_PROVIDERS: Final = tuple(provider for provider in _PROVIDERS if provider != "bedrock_converse")
_SITES: Final[tuple[Site, ...]] = ("tool", "system", "user_part", "tool_call", "tool_result", "string_message")
_FIVE_SITES: Final[tuple[Site, ...]] = ("tool", "system", "user_part", "tool_call", "string_message")
_EXPLICIT: Final[dict[str, JsonValue]] = {"mode": "explicit"}
_EXPLICIT_30M: Final[dict[str, JsonValue]] = {"mode": "explicit", "ttl": "30m"}
_BREAKPOINT: Final[Mapping[str, JsonValue]] = MappingProxyType({"prompt_cache_breakpoint": _EXPLICIT})
_NATIVE: Final[Mapping[str, JsonValue]] = MappingProxyType({"cache_control": EPHEMERAL})
_CITY: Final = CITIES[0]
_TOOL_RESULT_TEXT: Final = "sunny."
_BEDROCK_TOKEN: Final = "synthetic-bedrock-bearer"
_VERTEX_PROJECT: Final = "scripted-project"
_VERTEX_LOCATION: Final = "us-east5"
_TARGETS: Final[Mapping[Provider, str]] = MappingProxyType(
{
"anthropic": "/v1/messages",
"bedrock_converse": f"/model/{BEDROCK_MODEL}/converse",
"bedrock_invoke": f"/model/{BEDROCK_MODEL}/invoke",
"bedrock_claude_platform": "/v1/messages",
"vertex_ai": (
f"/v1/projects/{_VERTEX_PROJECT}/locations/{_VERTEX_LOCATION}"
f"/publishers/anthropic/models/{ANTHROPIC_MODEL}:rawPredict"
),
"azure_ai": "/anthropic/v1/messages",
}
)
_CAP_HINT: Final = "Claude accepts at most 4 cache breakpoints, found 5"
class _Nested(BaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
type: str
text: str | None = None
cache_control: JsonValue = None
class _Block(BaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
type: str
tool_use_id: str | None = None
content: str | tuple[_Nested, ...] | None = None
class _Message(BaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
role: str
content: str | tuple[_Block, ...]
class _Body(BaseModel):
model_config = ConfigDict(extra="allow", frozen=True)
messages: tuple[_Message, ...] = ()
def _nested_blocks(body: _Body) -> Iterator[tuple[_Block, _Nested]]:
for message in body.messages:
for block in message.content if isinstance(message.content, tuple) else ():
if isinstance(block.content, tuple):
yield from ((block, inner) for inner in block.content)
def _nested_tool_result_labels(request: Request) -> list[str]:
return [
f"user:tool_result:{block.tool_use_id}:{inner.type}:{inner.text}"
for block, inner in _nested_blocks(_Body.model_validate_json(request.body))
if inner.cache_control is not None
]
def _labels(provider: Provider, request: Request) -> list[str]:
if provider == "bedrock_converse":
return bedrock_labels(request)
return [*anthropic_labels(request), *_nested_tool_result_labels(request)]
def _respond(provider: Provider, request: Request) -> Reply:
return bedrock_peer(request) if provider == "bedrock_converse" else anthropic_peer(request)
def _labels(provider: Provider, request: Request) -> list[str]:
return bedrock_labels(request) if provider == "bedrock_converse" else anthropic_labels(request)
def _litellm_params(gateway: Gateway, wire: Wire, provider: Provider) -> dict[str, JsonValue]:
match provider:
case "anthropic":
@ -184,11 +128,7 @@ def _messages(marker: str, marks: Mapping[Site, Mapping[str, JsonValue]]) -> lis
{"role": "system", "content": [{"type": "text", "text": SYSTEM, **at("system")}]},
{"role": "user", "content": [{"type": "text", "text": ASK, **at("user_part")}]},
{"role": "assistant", "content": "", "tool_calls": [tool_call(_CITY, **at("tool_call"))]},
{
"role": "tool",
"tool_call_id": call_id(_CITY),
"content": [{"type": "text", "text": _TOOL_RESULT_TEXT, **at("tool_result_part")}],
},
{"role": "tool", "tool_call_id": call_id(_CITY), "content": "sunny.", **at("tool_result")},
{"role": "user", "content": final_text(marker), **at("string_message")},
]
@ -216,10 +156,8 @@ def _expected(provider: Provider, site: Site, marker: str) -> str:
return ASK_LABEL
case "tool_call":
return f"assistant:toolUse:{call_id(_CITY)}" if converse else tool_use_label(_CITY)
case "tool_result_part":
if converse:
return f"user:toolResult:{call_id(_CITY)}"
return f"user:tool_result:{call_id(_CITY)}:text:{_TOOL_RESULT_TEXT}"
case "tool_result":
return f"user:toolResult:{call_id(_CITY)}" if converse else f"user:tool_result:{call_id(_CITY)}"
case "string_message":
return final_label(marker)
case _:
@ -236,6 +174,11 @@ def _only_request(wire: Wire) -> Request:
return received[0]
def _automatic_mark(request: Request) -> JsonValue:
body: Final = json.loads(request.body)
return body.get("cache_control") if isinstance(body, dict) else None
@pytest.mark.parametrize("site", _SITES)
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_breakpoint_becomes_a_cache_mark_on_the_same_block(gateway: Gateway, provider: Provider, site: Site) -> None:
@ -245,50 +188,67 @@ def test_breakpoint_becomes_a_cache_mark_on_the_same_block(gateway: Gateway, pro
status, _, text = post_chat(gateway, _body(model, marker, _marked((site,))))
assert status == 200, text
request: Final = _only_request(wire)
assert request.target == _TARGETS[provider], request.target
assert b"prompt_cache_breakpoint" not in request.body, request.body
assert _labels(provider, request) == [_expected(provider, site, marker)]
def test_breakpoint_ttl_is_not_forwarded(gateway: Gateway) -> None:
def test_breakpoint_ttl_claude_supports_is_kept(gateway: Gateway) -> None:
marker: Final = new_marker()
marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": _EXPLICIT_30M})
with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "anthropic")
status, _, text = post_chat(gateway, _body(model, marker, marks))
assert status == 200, text
request: Final = _only_request(wire)
assert anthropic_marks(request) == (Mark(ASK_LABEL, None),)
assert b"30m" not in request.body, request.body
def test_a_client_cache_control_on_the_same_block_wins_over_the_breakpoint(gateway: Gateway) -> None:
marker: Final = new_marker()
hour: Final[dict[str, JsonValue]] = {"type": "ephemeral", "ttl": "1h"}
marks: Final = _marked(("user_part",), {"cache_control": hour, "prompt_cache_breakpoint": _EXPLICIT_30M})
marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}})
with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "anthropic")
status, _, text = post_chat(gateway, _body(model, marker, marks))
assert status == 200, text
request: Final = _only_request(wire)
assert anthropic_marks(request) == (Mark(ASK_LABEL, "1h"),)
assert b"prompt_cache_breakpoint" not in request.body, request.body
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_a_block_carrying_both_keys_keeps_one_cache_mark_and_loses_the_breakpoint(
gateway: Gateway, provider: Provider
) -> None:
def test_breakpoint_ttl_claude_lacks_is_a_400_before_any_provider_call(gateway: Gateway, provider: Provider) -> None:
marker: Final = new_marker()
marks: Final = _marked(("user_part",), {"cache_control": EPHEMERAL, "prompt_cache_breakpoint": _EXPLICIT_30M})
marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}})
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(gateway, _body(model, marker, marks))
received: Final = wire.drain()
assert status == 400, text
assert "30m" in text and "drop_params" in text, text
assert received == (), [request.target for request in received]
def test_breakpoint_ttl_claude_lacks_keeps_the_mark_under_drop_params(gateway: Gateway) -> None:
marker: Final = new_marker()
marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}})
with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "anthropic", drop_params=True)
status, _, text = post_chat(gateway, _body(model, marker, marks))
assert status == 200, text
request: Final = _only_request(wire)
assert _labels(provider, request) == [_expected(provider, "user_part", marker)]
assert b"prompt_cache_breakpoint" not in request.body, request.body
assert b"30m" not in request.body, request.body
assert anthropic_marks(request) == (Mark(ASK_LABEL, None),)
def test_malformed_breakpoint_is_a_400_before_any_provider_call(gateway: Gateway) -> None:
marker: Final = new_marker()
marks: Final = _marked(("user_part",), {"prompt_cache_breakpoint": {"mode": "sometimes"}})
with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "anthropic")
status, _, text = post_chat(gateway, _body(model, marker, marks))
received: Final = wire.drain()
assert status == 400, text
assert "prompt_cache_breakpoint" in text and "drop_params" in text, text
assert received == (), [request.target for request in received]
def test_a_client_cache_control_on_the_same_block_wins_over_the_breakpoint(gateway: Gateway) -> None:
marker: Final = new_marker()
hour: Final[dict[str, JsonValue]] = {"type": "ephemeral", "ttl": "1h"}
marks: Final = _marked(("user_part",), {"cache_control": hour, "prompt_cache_breakpoint": _EXPLICIT})
with wire_server(anthropic_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "anthropic")
status, _, text = post_chat(gateway, _body(model, marker, marks))
assert status == 200, text
request: Final = _only_request(wire)
assert anthropic_marks(request) == (Mark(ASK_LABEL, "1h"),)
@pytest.mark.parametrize("source", ("deployment", "request"))
@ -307,38 +267,19 @@ def test_five_breakpoints_keep_the_latest_four_under_drop_params(
assert _labels(provider, request) == _latest_four(provider, marker)
@pytest.mark.parametrize("mark", (_BREAKPOINT, _NATIVE), ids=("breakpoint", "cache_control"))
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_five_breakpoints_are_a_400_before_any_provider_call_without_drop_params(
gateway: Gateway, provider: Provider
def test_five_marks_without_drop_params_reach_the_provider_and_its_error_reaches_the_caller(
gateway: Gateway, provider: Provider, mark: Mapping[str, JsonValue]
) -> None:
marker: Final = new_marker()
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(gateway, _body(model, marker, _marked(_FIVE_SITES)))
status, _, text = post_chat(gateway, _body(model, marker, _marked(_FIVE_SITES, mark)))
received: Final = wire.drain()
assert status == 400, text
assert _CAP_HINT in text, text
assert "drop_params" in text, text
assert received == (), [request.target for request in received]
def test_a_tool_result_with_two_marked_parts_is_one_converse_cache_point_within_the_cap(gateway: Gateway) -> None:
marker: Final = new_marker()
marked_tool_result: Final[dict[str, JsonValue]] = {
"role": "tool",
"tool_call_id": call_id(_CITY),
"content": [{"type": "text", "text": text, **_BREAKPOINT} for text in (_TOOL_RESULT_TEXT, "warm.")],
}
messages: Final = _messages(marker, _marked(("system", "user_part", "string_message")))
with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "bedrock_converse")
status, _, text = post_chat(gateway, chat_body(model, [*messages[:3], marked_tool_result, *messages[4:]]))
assert status == 200, text
request: Final = _only_request(wire)
assert bedrock_labels(request) == [
_expected("bedrock_converse", site, marker)
for site in ("system", "user_part", "tool_result_part", "string_message")
]
assert "Found 5" in text, text
assert [len(_labels(provider, request)) for request in received] == [5], [request.target for request in received]
def test_injection_points_stand_down_behind_the_latest_four_breakpoints(gateway: Gateway) -> None:
@ -353,59 +294,25 @@ def test_injection_points_stand_down_behind_the_latest_four_breakpoints(gateway:
assert gateway_injected(response_id) is False
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_prompt_cache_options_are_a_400_before_any_provider_call_without_drop_params(
gateway: Gateway, provider: Provider
) -> None:
@pytest.mark.parametrize("provider", _AUTOMATIC_CACHING_PROVIDERS)
def test_implicit_prompt_cache_options_turn_on_claude_automatic_caching(gateway: Gateway, provider: Provider) -> None:
marker: Final = new_marker()
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(
gateway, _body(model, marker, _marked(("user_part",)), prompt_cache_options=_EXPLICIT_30M)
)
received: Final = wire.drain()
assert status == 400, text
assert "prompt_cache_options" in text, text
assert "drop_params" in text, text
assert received == (), [request.target for request in received]
@pytest.mark.parametrize("provider", _EXTRA_BODY_PROVIDERS)
def test_prompt_cache_options_in_extra_body_are_a_400_before_any_provider_call_without_drop_params(
gateway: Gateway, provider: Provider
) -> None:
marker: Final = new_marker()
extra_body: Final[dict[str, JsonValue]] = {"prompt_cache_options": _EXPLICIT}
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(gateway, _body(model, marker, _marked(("user_part",)), extra_body=extra_body))
received: Final = wire.drain()
assert status == 400, text
assert "prompt_cache_options" in text, text
assert "drop_params" in text, text
assert received == (), [request.target for request in received]
@pytest.mark.parametrize("provider", _EXTRA_BODY_PROVIDERS)
def test_prompt_cache_options_in_extra_body_are_dropped_under_drop_params(gateway: Gateway, provider: Provider) -> None:
marker: Final = new_marker()
extra_body: Final[dict[str, JsonValue]] = {"prompt_cache_options": _EXPLICIT}
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider, drop_params=True)
status, _, text = post_chat(gateway, _body(model, marker, _marked(("user_part",)), extra_body=extra_body))
status, _, text = post_chat(gateway, _body(model, marker, {}, prompt_cache_options={"mode": "implicit"}))
assert status == 200, text
request: Final = _only_request(wire)
assert b"prompt_cache_options" not in request.body, request.body
assert _labels(provider, request) == [ASK_LABEL]
assert _automatic_mark(request) == EPHEMERAL
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_prompt_cache_options_are_dropped_under_drop_params(gateway: Gateway, provider: Provider) -> None:
def test_explicit_prompt_cache_options_leave_only_the_breakpoints(gateway: Gateway, provider: Provider) -> None:
marker: Final = new_marker()
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider, drop_params=True)
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(
gateway, _body(model, marker, _marked(("user_part",)), prompt_cache_options=_EXPLICIT_30M)
gateway, _body(model, marker, _marked(("user_part",)), prompt_cache_options=_EXPLICIT)
)
assert status == 200, text
request: Final = _only_request(wire)
@ -413,19 +320,28 @@ def test_prompt_cache_options_are_dropped_under_drop_params(gateway: Gateway, pr
assert _labels(provider, request) == [ASK_LABEL]
@pytest.mark.parametrize("provider", _PROVIDERS)
def test_five_native_cache_control_marks_are_forwarded_and_the_provider_error_reaches_the_caller(
gateway: Gateway, provider: Provider
) -> None:
def test_implicit_prompt_cache_options_are_a_400_where_claude_has_no_automatic_caching(gateway: Gateway) -> None:
marker: Final = new_marker()
with wire_server(lambda request: _respond(provider, request)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, provider)
status, _, text = post_chat(gateway, _body(model, marker, _marked(_FIVE_SITES, _NATIVE)))
with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "bedrock_converse")
status, _, text = post_chat(gateway, _body(model, marker, {}, prompt_cache_options={"mode": "implicit"}))
received: Final = wire.drain()
assert status == 400, text
assert "Found 5" in text, text
assert _CAP_HINT not in text, text
assert [len(_labels(provider, request)) for request in received] == [5], [request.target for request in received]
assert "prompt_cache_options" in text and "drop_params" in text, text
assert received == (), [request.target for request in received]
def test_implicit_prompt_cache_options_are_dropped_where_claude_has_no_automatic_caching_under_drop_params(
gateway: Gateway,
) -> None:
marker: Final = new_marker()
with wire_server(bedrock_peer) as wire, gateway.scenario() as scenario:
model: Final = _deployment(gateway, scenario, wire, "bedrock_converse", drop_params=True)
status, _, text = post_chat(gateway, _body(model, marker, {}, prompt_cache_options={"mode": "implicit"}))
assert status == 200, text
request: Final = _only_request(wire)
assert b"prompt_cache_options" not in request.body, request.body
assert bedrock_labels(request) == []
def _drop_params_config(directory: Path) -> Path:
@ -434,7 +350,7 @@ def _drop_params_config(directory: Path) -> Path:
@pytest.fixture(scope="module")
def drop_params_gateway(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Gateway]:
"""A proxy running with `litellm_settings: drop_params: true`, the setting the cap error names"""
"""A proxy running with `litellm_settings: drop_params: true`"""
with gateway_from_environment() as lane:
directory: Final = tmp_path_factory.mktemp("drop_params")
with owned_proxy(lane, directory, {}, config=_drop_params_config(directory)) as owned:

132
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py
View file

@ -23,18 +23,12 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
hoist_images_from_tool_messages,
is_encrypted_reasoning_block,
merge_consecutive_system_messages,
message_with_prompt_cache_breakpoints_as_cache_control,
message_without_cache_control_marks,
messages_with_prompt_cache_breakpoints_as_cache_control,
parse_tool_call_arguments,
prompt_cache_breakpoint_as_cache_control,
responses_reasoning_items_from_thinking_blocks,
salvage_concatenated_tool_arguments,
split_concatenated_json_objects,
strip_encrypted_reasoning_from_messages,
system_messages_first,
tool_with_prompt_cache_breakpoint_as_cache_control,
tools_with_prompt_cache_breakpoints_as_cache_control,
update_messages_with_model_file_ids,
with_prompt_cache_breakpoint,
)
@ -2312,129 +2306,3 @@ def test_a_breakpoint_the_client_wrote_survives_a_per_part_cache_control() -> No
"text": "rules",
"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"},
}
_EPHEMERAL: Final = {"type": "ephemeral"}
_EXPLICIT: Final = {"mode": "explicit"}
_EXPLICIT_30M: Final = {"mode": "explicit", "ttl": "30m"}
def _lookup_call(identity: str, **fields: object) -> dict[str, object]:
return {"id": identity, "type": "function", "function": {"name": "lookup", "arguments": "{}"}, **fields}
def test_prompt_cache_breakpoint_becomes_an_ephemeral_cache_control_without_the_ttl() -> None:
marked: Final = {"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT_30M}
assert prompt_cache_breakpoint_as_cache_control(marked) == {
"type": "text",
"text": "rules",
"cache_control": _EPHEMERAL,
}
def test_a_cache_control_the_client_wrote_wins_over_the_breakpoint_on_the_same_block() -> None:
both: Final = {
"type": "text",
"text": "rules",
"cache_control": {"type": "ephemeral", "ttl": "1h"},
"prompt_cache_breakpoint": _EXPLICIT_30M,
}
assert prompt_cache_breakpoint_as_cache_control(both) == {
"type": "text",
"text": "rules",
"cache_control": {"type": "ephemeral", "ttl": "1h"},
}
def test_a_block_without_a_breakpoint_is_returned_untouched() -> None:
unmarked: Final = {"type": "text", "text": "rules", "cache_control": _EPHEMERAL}
assert prompt_cache_breakpoint_as_cache_control(unmarked) is unmarked
def test_breakpoints_are_renamed_at_the_message_level_on_parts_and_on_tool_calls() -> None:
message: Final = {
"role": "assistant",
"content": [{"type": "text", "text": "a", "prompt_cache_breakpoint": _EXPLICIT}, {"type": "text", "text": "b"}],
"tool_calls": [_lookup_call("call_1", prompt_cache_breakpoint=_EXPLICIT_30M), _lookup_call("call_2")],
"prompt_cache_breakpoint": _EXPLICIT,
}
assert message_with_prompt_cache_breakpoints_as_cache_control(message) == {
"role": "assistant",
"content": [{"type": "text", "text": "a", "cache_control": _EPHEMERAL}, {"type": "text", "text": "b"}],
"tool_calls": [_lookup_call("call_1", cache_control=_EPHEMERAL), _lookup_call("call_2")],
"cache_control": _EPHEMERAL,
}
def test_a_message_level_breakpoint_on_string_content_is_renamed_and_other_messages_are_kept() -> None:
unmarked: Final = {"role": "user", "content": "unmarked"}
renamed: Final = messages_with_prompt_cache_breakpoints_as_cache_control(
[{"role": "user", "content": "hi", "prompt_cache_breakpoint": _EXPLICIT}, unmarked]
)
assert renamed == ({"role": "user", "content": "hi", "cache_control": _EPHEMERAL}, unmarked)
assert renamed[1] is unmarked
@pytest.mark.parametrize(
"tool",
[
{"type": "function", "function": {"name": "lookup", "parameters": {}}, "prompt_cache_breakpoint": _EXPLICIT},
{"type": "function", "function": {"name": "lookup", "parameters": {}, "prompt_cache_breakpoint": _EXPLICIT}},
],
ids=["top-level", "under-function"],
)
def test_a_tool_breakpoint_becomes_a_top_level_cache_control(tool: dict[str, object]) -> None:
assert tool_with_prompt_cache_breakpoint_as_cache_control(tool) == {
"type": "function",
"function": {"name": "lookup", "parameters": {}},
"cache_control": _EPHEMERAL,
}
def test_a_tool_cache_control_the_client_wrote_wins_over_the_breakpoint() -> None:
tool: Final = {
"type": "function",
"function": {
"name": "lookup",
"cache_control": {"type": "ephemeral", "ttl": "1h"},
"prompt_cache_breakpoint": _EXPLICIT_30M,
},
}
assert tool_with_prompt_cache_breakpoint_as_cache_control(tool) == {
"type": "function",
"function": {"name": "lookup", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
}
def test_tools_that_are_not_a_list_carry_no_marks_and_unmarked_tools_are_kept() -> None:
unmarked: Final = {"type": "function", "function": {"name": "lookup"}}
assert tools_with_prompt_cache_breakpoints_as_cache_control(None) == ()
assert tools_with_prompt_cache_breakpoints_as_cache_control([unmarked, "opaque"]) == (unmarked, "opaque")
def test_message_without_cache_control_marks_strips_only_the_listed_positions() -> None:
message: Final = {
"role": "assistant",
"content": [
{"type": "text", "text": "a", "cache_control": _EPHEMERAL},
{"type": "text", "text": "b", "cache_control": _EPHEMERAL},
],
"tool_calls": [_lookup_call("call_1", cache_control=_EPHEMERAL), _lookup_call("call_2", cache_control=_EPHEMERAL)],
"cache_control": _EPHEMERAL,
}
assert message_without_cache_control_marks(
message, message_mark=True, parts=frozenset({0}), tool_calls=frozenset({1})
) == {
"role": "assistant",
"content": [{"type": "text", "text": "a"}, {"type": "text", "text": "b", "cache_control": _EPHEMERAL}],
"tool_calls": [_lookup_call("call_1", cache_control=_EPHEMERAL), _lookup_call("call_2")],
}

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

@ -1,6 +1,5 @@
import copy
import json
from collections.abc import Callable
from typing import Final
from unittest.mock import MagicMock, patch
@ -8736,369 +8735,3 @@ def test_completion_surfaces_response_citations_in_provider_specific_fields(resp
assert response.choices[0].message.provider_specific_fields["citations"] == [
[{**citation, "supported_text": "the grass is green"}]
]
EXPLICIT: Final = {"mode": "explicit"}
EXPLICIT_30M: Final = {"mode": "explicit", "ttl": "30m"}
_WEATHER_FUNCTION: Final = {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"parameters": WEATHER_PARAMETERS,
}
_ANTHROPIC_WEATHER_TOOL: Final = {
"name": "get_current_weather",
"description": "Get the current weather in a given location",
"input_schema": WEATHER_PARAMETERS,
"type": "custom",
}
_ANTHROPIC_REPLY: Final = {
"id": "msg_breakpoints",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "Hello!"}],
"model": PROMPT_CACHING_MODEL,
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 12, "output_tokens": 6},
}
def _five_marked_parts() -> list[dict[str, object]]:
return [{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": EXPLICIT} for index in range(5)]
def _latest_four_marked() -> list[dict[str, object]]:
return [
{"type": "text", "text": "part 0"},
*({"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(1, 5)),
]
async def test_prompt_cache_breakpoints_on_system_user_and_tool_definition_become_cache_control(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[
{"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": EXPLICIT}]},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": EXPLICIT_30M}]},
],
tools=[{"type": "function", "function": {**_WEATHER_FUNCTION, "prompt_cache_breakpoint": EXPLICIT}}],
)
assert json.loads(request.content) == {
"model": PROMPT_CACHING_MODEL,
"system": [{"type": "text", "text": "rules", "cache_control": EPHEMERAL}],
"messages": [{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": EPHEMERAL}]}],
"tools": [{**_ANTHROPIC_WEATHER_TOOL, "cache_control": EPHEMERAL}],
"max_tokens": litellm.get_max_tokens(PROMPT_CACHING_MODEL),
}
async def test_prompt_cache_breakpoints_on_tool_calls_tool_results_and_string_messages_become_cache_control(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_current_weather", "arguments": '{"location": "Boston"}'},
"prompt_cache_breakpoint": EXPLICIT,
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [{"type": "text", "text": "sunny", "prompt_cache_breakpoint": EXPLICIT}],
},
{"role": "assistant", "content": "It is sunny."},
{"role": "user", "content": "thanks", "prompt_cache_breakpoint": EXPLICIT_30M},
],
tools=[{"type": "function", "function": _WEATHER_FUNCTION}],
)
assert "prompt_cache_breakpoint" not in request.content.decode()
assert json.loads(request.content)["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "weather?"}]},
{
"role": "assistant",
"content": [
{
"type": "tool_use",
"id": "call_1",
"name": "get_current_weather",
"input": {"location": "Boston"},
"cache_control": EPHEMERAL,
}
],
},
{
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "call_1",
"content": [{"type": "text", "text": "sunny", "cache_control": EPHEMERAL}],
}
],
},
{"role": "assistant", "content": [{"type": "text", "text": "It is sunny."}]},
{"role": "user", "content": [{"type": "text", "text": "thanks", "cache_control": EPHEMERAL}]},
]
async def test_a_cache_control_the_client_wrote_wins_over_the_breakpoint_on_the_same_block(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[
{
"role": "user",
"content": [
{
"type": "text",
"text": "hi",
"cache_control": {"type": "ephemeral", "ttl": "1h"},
"prompt_cache_breakpoint": EXPLICIT_30M,
}
],
}
],
)
assert "prompt_cache_breakpoint" not in request.content.decode()
assert json.loads(request.content)["messages"] == [
{
"role": "user",
"content": [{"type": "text", "text": "hi", "cache_control": {"type": "ephemeral", "ttl": "1h"}}],
}
]
@pytest.mark.parametrize("drop_params_source", ["kwarg", "global"])
async def test_five_breakpoints_keep_the_latest_four_under_drop_params(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, drop_params_source: str
) -> None:
monkeypatch.setattr(litellm, "drop_params", drop_params_source == "global")
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[{"role": "user", "content": _five_marked_parts()}],
**({"drop_params": True} if drop_params_source == "kwarg" else {}),
)
assert json.loads(request.content)["messages"] == [{"role": "user", "content": _latest_four_marked()}]
async def test_five_breakpoints_are_a_bad_request_before_any_provider_call_without_drop_params(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(200, json=_ANTHROPIC_REPLY)
)
with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 5") as caught:
await litellm.acompletion(
api_key="mock_api_key",
model=f"anthropic/{PROMPT_CACHING_MODEL}",
messages=[{"role": "user", "content": _five_marked_parts()}],
num_retries=0,
)
assert "drop_params=True" in str(caught.value)
assert route.call_count == 0
async def test_configured_injection_points_stand_down_when_five_breakpoints_are_cut_to_four(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[{"role": "system", "content": "rules"}, {"role": "user", "content": _five_marked_parts()}],
drop_params=True,
cache_control_injection_points=[
{"location": "message", "role": "system"},
{"location": "message", "index": -1},
],
)
body: Final = json.loads(request.content)
assert body["system"] == [{"type": "text", "text": "rules"}]
assert body["messages"] == [{"role": "user", "content": _latest_four_marked()}]
async def test_prompt_cache_options_are_rejected_before_any_provider_call_without_drop_params(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(200, json=_ANTHROPIC_REPLY)
)
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
await litellm.acompletion(
api_key="mock_api_key",
model=f"anthropic/{PROMPT_CACHING_MODEL}",
messages=[{"role": "user", "content": "hi"}],
prompt_cache_options={"mode": "explicit", "ttl": "30m"},
num_retries=0,
)
assert route.call_count == 0
async def test_prompt_cache_options_are_dropped_under_drop_params(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[{"role": "user", "content": "hi"}],
prompt_cache_options={"mode": "explicit", "ttl": "30m"},
drop_params=True,
)
assert json.loads(request.content) == {
"model": PROMPT_CACHING_MODEL,
"messages": [{"role": "user", "content": [{"type": "text", "text": "hi"}]}],
"max_tokens": litellm.get_max_tokens(PROMPT_CACHING_MODEL),
}
async def test_five_native_cache_control_marks_are_forwarded_and_the_provider_answer_reaches_the_caller(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", True)
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
marked: Final = [{"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(5)]
route: Final = respx_mock.post("https://api.anthropic.com/v1/messages").mock(
return_value=httpx.Response(
400,
json={
"type": "error",
"error": {
"type": "invalid_request_error",
"message": "A maximum of 4 blocks with cache_control may be provided. Found 5.",
},
},
)
)
with pytest.raises(litellm.BadRequestError, match=r"Found 5"):
await litellm.acompletion(
api_key="mock_api_key",
model=f"anthropic/{PROMPT_CACHING_MODEL}",
messages=[{"role": "user", "content": marked}],
num_retries=0,
)
assert route.call_count == 1
assert json.loads(route.calls.last.request.content)["messages"] == [{"role": "user", "content": marked}]
async def test_a_breakpoint_only_on_a_tool_definition_still_holds_the_request_to_four_marks(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
marked: Final = [{"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(4)]
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[{"role": "user", "content": marked}],
tools=[{"type": "function", "function": _WEATHER_FUNCTION, "prompt_cache_breakpoint": EXPLICIT}],
drop_params=True,
)
body: Final = json.loads(request.content)
assert body["tools"] == [_ANTHROPIC_WEATHER_TOOL]
assert body["messages"] == [{"role": "user", "content": marked}]
async def test_the_automatic_cache_control_takes_one_of_the_four_slots(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": EXPLICIT} for index in range(4)
],
}
],
cache_control=EPHEMERAL,
drop_params=True,
)
body: Final = json.loads(request.content)
assert body["cache_control"] == EPHEMERAL
assert body["messages"] == [
{
"role": "user",
"content": [
{"type": "text", "text": "part 0"},
*({"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(1, 4)),
],
}
]
async def test_a_message_level_breakpoint_on_list_content_does_not_count_toward_the_cap(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
request: Final = await _send_prompt_caching_request(
respx_mock,
monkeypatch,
messages=[
{
"role": "user",
"content": [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": EXPLICIT} for index in range(4)
],
"prompt_cache_breakpoint": EXPLICIT,
}
],
)
assert json.loads(request.content)["messages"] == [
{
"role": "user",
"content": [{"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(4)],
}
]
@pytest.mark.parametrize(
"tools",
[
pytest.param(lambda tool: (tool,), id="tuple"),
pytest.param(lambda tool: iter([tool]), id="iterator"),
],
)
def test_map_openai_params_maps_tools_given_as_any_iterable(tools: Callable[[object], object]) -> None:
optional_params: Final = AnthropicConfig().map_openai_params(
non_default_params={"tools": tools({"type": "function", "function": _WEATHER_FUNCTION})},
optional_params={},
model=PROMPT_CACHING_MODEL,
drop_params=False,
)
assert optional_params["tools"] == [_ANTHROPIC_WEATHER_TOOL]

512
tests/unit/llms/anthropic/chat/test_prompt_cache_breakpoints.py
View file

@ -2,409 +2,201 @@ from typing import Final
import pytest
import litellm
from litellm.exceptions import UnsupportedParamsError
from litellm.llms.anthropic.chat.prompt_cache_breakpoints import (
CacheBreakpointCapExceeded,
ClaudeCacheMarks,
PromptCacheOptionsUnsupported,
claude_cache_marks,
claude_extra_body,
drop_params_enabled,
resolved_claude_cache_marks,
resolved_claude_extra_body,
tools_carry_prompt_cache_breakpoint,
ClaudePromptCache,
UnsupportedPromptCacheParam,
claude_prompt_cache,
resolved_claude_prompt_cache,
)
EPHEMERAL: Final = {"type": "ephemeral"}
EXPLICIT: Final = {"mode": "explicit"}
EXPLICIT_30M: Final = {"mode": "explicit", "ttl": "30m"}
MODEL: Final = "claude-opus-5-5"
_EXPLICIT: Final = {"mode": "explicit"}
_EPHEMERAL: Final = {"type": "ephemeral"}
def _text(text: str, **fields: object) -> dict[str, object]:
return {"type": "text", "text": text, **fields}
def _cache(
messages: list[object],
tools: list[object] | None = None,
params: dict[str, object] | None = None,
*,
accepts_automatic_caching: bool = True,
drop_params: bool = False,
) -> ClaudePromptCache | UnsupportedPromptCacheParam:
return claude_prompt_cache(
messages,
tools,
params or {},
accepts_automatic_caching=accepts_automatic_caching,
drop_params=drop_params,
)
def _tool(**fields: object) -> dict[str, object]:
return {"type": "function", "function": {"name": "lookup", "parameters": {}}, **fields}
def _resolved(
messages: list[object],
tools: list[object] | None = None,
params: dict[str, object] | None = None,
*,
accepts_automatic_caching: bool = True,
drop_params: bool = False,
) -> ClaudePromptCache:
outcome: Final = _cache(
messages, tools, params, accepts_automatic_caching=accepts_automatic_caching, drop_params=drop_params
)
assert isinstance(outcome, ClaudePromptCache), outcome
return outcome
def _call(identity: str, **fields: object) -> dict[str, object]:
return {"id": identity, "type": "function", "function": {"name": "lookup", "arguments": "{}"}, **fields}
def _part(text: str, **marks: object) -> dict[str, object]:
return {"type": "text", "text": text, **marks}
def _six_breakpoints() -> tuple[list[dict[str, object]], list[dict[str, object]]]:
"""A tool definition, a system part, a user part, a tool call, a tool result part and a string message, every
one marked, in the order Claude reads them"""
def test_breakpoints_become_cache_control_on_every_block_that_carries_one():
messages: Final = [
{"role": "system", "content": [_text("rules", prompt_cache_breakpoint=EXPLICIT)]},
{"role": "user", "content": [_text("hi", prompt_cache_breakpoint=EXPLICIT_30M)]},
{"role": "assistant", "content": "", "tool_calls": [_call("call_1", prompt_cache_breakpoint=EXPLICIT)]},
{"role": "tool", "tool_call_id": "call_1", "content": [_text("sunny", prompt_cache_breakpoint=EXPLICIT)]},
{"role": "user", "content": "done", "prompt_cache_breakpoint": EXPLICIT},
]
return messages, [_tool(prompt_cache_breakpoint=EXPLICIT)]
def test_drop_params_keeps_the_latest_four_marks_in_wire_order() -> None:
messages, tools = _six_breakpoints()
marks: Final = claude_cache_marks(messages, tools, automatic_breakpoint=False, drop_params=True)
assert marks == ClaudeCacheMarks(
messages=(
{"role": "system", "content": [_text("rules")]},
{"role": "user", "content": [_text("hi", cache_control=EPHEMERAL)]},
{"role": "assistant", "content": "", "tool_calls": [_call("call_1", cache_control=EPHEMERAL)]},
{"role": "tool", "tool_call_id": "call_1", "content": [_text("sunny", cache_control=EPHEMERAL)]},
{"role": "user", "content": "done", "cache_control": EPHEMERAL},
),
tools=(_tool(),),
)
def test_four_breakpoints_are_renamed_without_dropping_anything() -> None:
messages, tools = _six_breakpoints()
marks: Final = claude_cache_marks(messages[2:], tools, automatic_breakpoint=False, drop_params=False)
assert marks == ClaudeCacheMarks(
messages=(
{"role": "assistant", "content": "", "tool_calls": [_call("call_1", cache_control=EPHEMERAL)]},
{"role": "tool", "tool_call_id": "call_1", "content": [_text("sunny", cache_control=EPHEMERAL)]},
{"role": "user", "content": "done", "cache_control": EPHEMERAL},
),
tools=(_tool(cache_control=EPHEMERAL),),
)
def test_the_automatic_breakpoint_takes_a_slot_and_is_never_dropped() -> None:
parts: Final = [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]
marks: Final = claude_cache_marks(
[{"role": "user", "content": parts}], None, automatic_breakpoint=True, drop_params=True
)
assert marks == ClaudeCacheMarks(
messages=(
{
"role": "user",
"content": [
_text("part 0"),
*(_text(f"part {index}", cache_control=EPHEMERAL) for index in range(1, 4)),
],
},
),
tools=(),
)
def test_without_drop_params_the_excess_is_reported_with_the_count() -> None:
messages, tools = _six_breakpoints()
assert claude_cache_marks(messages, tools, automatic_breakpoint=False, drop_params=False) == (
CacheBreakpointCapExceeded(found=6)
)
def test_a_request_without_breakpoints_is_forwarded_as_written_even_past_the_cap() -> None:
messages: Final = [
{"role": "user", "content": [_text(f"part {index}", cache_control=EPHEMERAL) for index in range(5)]}
]
tools: Final = [_tool(cache_control=EPHEMERAL)]
marks: Final = claude_cache_marks(messages, tools, automatic_breakpoint=True, drop_params=True)
assert marks == ClaudeCacheMarks(messages=(messages[0],), tools=(tools[0],))
assert marks.messages[0] is messages[0]
def test_a_message_level_mark_on_list_content_takes_no_slot_because_claude_never_renders_it() -> None:
marked_parts: Final = [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(3)]
messages: Final = [
{"role": "user", "content": "first", "prompt_cache_breakpoint": EXPLICIT},
{"role": "user", "content": [*marked_parts, _text("tail")], "prompt_cache_breakpoint": EXPLICIT},
]
marks: Final = claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False)
assert marks == ClaudeCacheMarks(
messages=(
{"role": "user", "content": "first", "cache_control": EPHEMERAL},
{
"role": "user",
"content": [*(_text(f"part {index}", cache_control=EPHEMERAL) for index in range(3)), _text("tail")],
"cache_control": EPHEMERAL,
},
),
tools=(),
)
def test_a_message_level_mark_on_a_tool_result_with_list_content_takes_a_slot() -> None:
messages: Final = [
{"role": "user", "content": [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [_text("sunny")],
"prompt_cache_breakpoint": EXPLICIT,
},
]
assert claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False) == (
CacheBreakpointCapExceeded(found=5)
)
def test_a_message_level_mark_on_a_function_result_with_list_content_takes_a_slot() -> None:
messages: Final = [
{"role": "user", "content": [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]},
{"role": "function", "name": "lookup", "content": [_text("sunny")], "prompt_cache_breakpoint": EXPLICIT},
]
assert claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False) == (
CacheBreakpointCapExceeded(found=5)
)
def _four_marked_parts_then(message: dict[str, object]) -> list[dict[str, object]]:
return [
{"role": "user", "content": [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]},
message,
]
def test_a_mark_on_an_empty_assistant_string_takes_no_anthropic_slot_but_a_converse_slot() -> None:
messages: Final = _four_marked_parts_then({"role": "assistant", "content": "", "prompt_cache_breakpoint": EXPLICIT})
anthropic: Final = claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False)
converse: Final = claude_cache_marks(
messages, None, automatic_breakpoint=False, drop_params=False, wire="bedrock_converse"
)
assert isinstance(anthropic, ClaudeCacheMarks), anthropic
assert converse == CacheBreakpointCapExceeded(found=5)
def test_a_mark_on_a_tool_call_rebuilt_as_server_tool_use_takes_no_anthropic_slot() -> None:
messages: Final = _four_marked_parts_then(
{"role": "system", "content": [_part("sys", prompt_cache_breakpoint=_EXPLICIT)]},
{"role": "user", "content": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "1h"}},
{
"role": "assistant",
"content": "searching",
"tool_calls": [_call("srvtoolu_web_search", prompt_cache_breakpoint=EXPLICIT)],
"provider_specific_fields": {"web_search_results": [{"tool_use_id": "srvtoolu_web_search"}]},
}
)
anthropic: Final = claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False)
converse: Final = claude_cache_marks(
messages, None, automatic_breakpoint=False, drop_params=False, wire="bedrock_converse"
)
assert isinstance(anthropic, ClaudeCacheMarks), anthropic
assert converse == CacheBreakpointCapExceeded(found=5)
def test_tools_already_mapped_to_the_anthropic_shape_still_arm_the_cap() -> None:
mapped_tools: Final = [
{"name": f"tool_{index}", "input_schema": {}, "cache_control": EPHEMERAL} for index in range(4)
]
messages: Final = [{"role": "user", "content": [_text("hi", prompt_cache_breakpoint=EXPLICIT)]}]
assert claude_cache_marks(
messages, mapped_tools, automatic_breakpoint=False, drop_params=False, tools_carried_breakpoint=True
) == CacheBreakpointCapExceeded(found=5)
assert claude_cache_marks(
messages, mapped_tools, automatic_breakpoint=False, drop_params=True, tools_carried_breakpoint=True
) == ClaudeCacheMarks(
messages=({"role": "user", "content": [_text("hi", cache_control=EPHEMERAL)]},),
tools=({"name": "tool_0", "input_schema": {}}, *mapped_tools[1:]),
)
def _tool_result_with_two_marked_parts() -> list[dict[str, object]]:
return [
{"role": "system", "content": [_text("rules", prompt_cache_breakpoint=EXPLICIT)]},
{"role": "user", "content": [_text("hi", prompt_cache_breakpoint=EXPLICIT)]},
{"role": "assistant", "content": "", "tool_calls": [_call("call_1")]},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [
_text("sunny", prompt_cache_breakpoint=EXPLICIT),
_text("warm", prompt_cache_breakpoint=EXPLICIT),
],
"content": None,
"tool_calls": [{"id": "c1", "type": "function", "prompt_cache_breakpoint": _EXPLICIT}],
},
{"role": "user", "content": "done", "prompt_cache_breakpoint": EXPLICIT},
]
tools: Final = [
{"type": "function", "function": {"name": "a"}, "prompt_cache_breakpoint": _EXPLICIT},
{"type": "function", "function": {"name": "b", "prompt_cache_breakpoint": _EXPLICIT}},
]
result: Final = _resolved(messages, tools)
def test_a_tool_result_with_two_marked_parts_takes_one_converse_slot_and_two_anthropic_slots() -> None:
messages: Final = _tool_result_with_two_marked_parts()
converse: Final = claude_cache_marks(
messages, None, automatic_breakpoint=False, drop_params=False, wire="bedrock_converse"
)
assert converse == ClaudeCacheMarks(
messages=(
{"role": "system", "content": [_text("rules", cache_control=EPHEMERAL)]},
{"role": "user", "content": [_text("hi", cache_control=EPHEMERAL)]},
{"role": "assistant", "content": "", "tool_calls": [_call("call_1")]},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [_text("sunny", cache_control=EPHEMERAL), _text("warm", cache_control=EPHEMERAL)],
},
{"role": "user", "content": "done", "cache_control": EPHEMERAL},
),
tools=(),
)
assert claude_cache_marks(messages, None, automatic_breakpoint=False, drop_params=False) == (
CacheBreakpointCapExceeded(found=5)
)
def test_dropping_a_converse_tool_result_slot_clears_every_mark_on_that_message() -> None:
messages: Final = [
assert list(result.messages) == [
{"role": "system", "content": [_part("sys", cache_control=_EPHEMERAL)]},
{"role": "user", "content": "hi", "cache_control": {"type": "ephemeral", "ttl": "1h"}},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [
_text("sunny", prompt_cache_breakpoint=EXPLICIT),
_text("warm", prompt_cache_breakpoint=EXPLICIT),
],
"prompt_cache_breakpoint": EXPLICIT,
"role": "assistant",
"content": None,
"tool_calls": [{"id": "c1", "type": "function", "cache_control": _EPHEMERAL}],
},
{"role": "user", "content": [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]},
]
assert list(result.tools or ()) == [
{"type": "function", "function": {"name": "a"}, "cache_control": _EPHEMERAL},
{"type": "function", "function": {"name": "b", "cache_control": _EPHEMERAL}},
]
marks: Final = claude_cache_marks(
messages, None, automatic_breakpoint=False, drop_params=True, wire="bedrock_converse"
)
assert marks == ClaudeCacheMarks(
messages=(
{"role": "tool", "tool_call_id": "call_1", "content": [_text("sunny"), _text("warm")]},
{"role": "user", "content": [_text(f"part {index}", cache_control=EPHEMERAL) for index in range(4)]},
),
tools=(),
)
def test_a_request_without_breakpoints_or_options_is_returned_as_written():
messages: Final = [{"role": "user", "content": [_part("hi", cache_control=_EPHEMERAL)] * 5}]
params: Final = {"temperature": 0.1}
result: Final = _resolved(messages, None, params, drop_params=True)
assert result == ClaudePromptCache(tuple(messages), None, params)
def test_a_function_level_cache_control_takes_no_converse_slot() -> None:
tools: Final = [{"type": "function", "function": {"name": "lookup", "parameters": {}, "cache_control": EPHEMERAL}}]
def test_a_client_cache_control_on_the_same_block_wins():
hour: Final = {"type": "ephemeral", "ttl": "1h"}
messages: Final = [
{"role": "user", "content": [_text(f"part {index}", prompt_cache_breakpoint=EXPLICIT) for index in range(4)]}
{"role": "user", "content": [_part("hi", cache_control=hour, prompt_cache_breakpoint={"ttl": "9d"})]}
]
converse: Final = claude_cache_marks(
messages, tools, automatic_breakpoint=False, drop_params=False, wire="bedrock_converse"
)
assert converse == ClaudeCacheMarks(
messages=(
{"role": "user", "content": [_text(f"part {index}", cache_control=EPHEMERAL) for index in range(4)]},
),
tools=(tools[0],),
)
assert claude_cache_marks(messages, tools, automatic_breakpoint=False, drop_params=False) == (
CacheBreakpointCapExceeded(found=5)
)
def test_the_cap_error_is_a_public_bad_request_naming_drop_params() -> None:
with pytest.raises(litellm.BadRequestError) as caught:
resolved_claude_cache_marks(CacheBreakpointCapExceeded(found=6), model=MODEL, llm_provider="anthropic")
assert "Claude accepts at most 4 cache breakpoints, found 6" in str(caught.value)
assert "drop_params=True" in str(caught.value)
assert caught.value.status_code == 400
assert caught.value.llm_provider == "anthropic"
def test_marks_within_the_cap_resolve_to_themselves() -> None:
marks: Final = ClaudeCacheMarks(messages=(), tools=())
assert resolved_claude_cache_marks(marks, model=MODEL, llm_provider="anthropic") is marks
def test_prompt_cache_options_are_reported_before_the_marks_unless_drop_params() -> None:
messages, tools = _six_breakpoints()
assert claude_cache_marks(
messages, tools, automatic_breakpoint=False, drop_params=False, prompt_cache_options=EXPLICIT
) == PromptCacheOptionsUnsupported(EXPLICIT)
kept: Final = claude_cache_marks(
messages, tools, automatic_breakpoint=False, drop_params=True, prompt_cache_options=EXPLICIT
)
assert isinstance(kept, ClaudeCacheMarks)
assert claude_cache_marks(messages, tools, automatic_breakpoint=False, drop_params=True) == kept
def test_the_prompt_cache_options_error_is_an_unsupported_param_naming_drop_params() -> None:
with pytest.raises(litellm.UnsupportedParamsError) as caught:
resolved_claude_cache_marks(PromptCacheOptionsUnsupported(EXPLICIT), model=MODEL, llm_provider="anthropic")
assert "prompt_cache_options" in str(caught.value)
assert "drop_params" in str(caught.value)
assert caught.value.status_code == 400
assert list(_resolved(messages).messages) == [{"role": "user", "content": [_part("hi", cache_control=hour)]}]
@pytest.mark.parametrize(
("extra_body", "drop_params", "expected"),
[
({"metadata": {"a": 1}}, False, {"metadata": {"a": 1}}),
({"metadata": {"a": 1}, "prompt_cache_options": EXPLICIT}, True, {"metadata": {"a": 1}}),
({"metadata": {"a": 1}, "prompt_cache_options": EXPLICIT}, False, PromptCacheOptionsUnsupported(EXPLICIT)),
],
ids=["no-options", "dropped", "rejected"],
"marker, kept",
[({"mode": "explicit", "ttl": "30m"}, _EPHEMERAL), ({"mode": "sometimes"}, None), ("yes", None)],
)
def test_extra_body_follows_the_prompt_cache_options_rule(
extra_body: dict[str, object], drop_params: bool, expected: object
) -> None:
assert claude_extra_body(extra_body, drop_params=drop_params) == expected
def test_a_breakpoint_claude_cannot_honor_is_reported_or_adjusted_under_drop_params(marker: object, kept: object):
messages: Final = [{"role": "user", "content": "hi", "prompt_cache_breakpoint": marker}]
problem: Final = _cache(messages)
dropped: Final = _resolved(messages, drop_params=True)
assert isinstance(problem, UnsupportedPromptCacheParam)
assert (problem.param, problem.value) == ("prompt_cache_breakpoint", marker)
assert list(dropped.messages) == [{"role": "user", "content": "hi", **({"cache_control": kept} if kept else {})}]
def test_a_rejected_extra_body_raises_and_an_accepted_one_resolves_to_itself() -> None:
body: Final = {"metadata": {"a": 1}}
def test_an_unsupported_breakpoint_raises_a_400_naming_drop_params():
problem: Final = _cache(
[{"role": "user", "content": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}}]
)
assert resolved_claude_extra_body(body, model=MODEL, llm_provider="anthropic") is body
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
resolved_claude_extra_body(PromptCacheOptionsUnsupported(EXPLICIT), model=MODEL, llm_provider="anthropic")
with pytest.raises(UnsupportedParamsError) as raised:
resolved_claude_prompt_cache(problem, model="claude-x", llm_provider="anthropic")
assert raised.value.status_code == 400
assert "30m" in raised.value.message and "drop_params" in raised.value.message
@pytest.mark.parametrize(
("global_drop", "litellm_params", "expected"),
[
(False, {}, False),
(True, {}, True),
(False, {"drop_params": True}, True),
(False, {"drop_params": "true"}, True),
(False, {"drop_params": False}, False),
],
ids=["off", "global", "request", "request-string", "request-false"],
"options, automatic",
[({"mode": "implicit"}, _EPHEMERAL), ({"mode": "implicit", "ttl": "1h"}, {"type": "ephemeral", "ttl": "1h"})],
)
def test_drop_params_enabled_reads_the_global_and_the_request(
global_drop: bool, litellm_params: dict[str, object], expected: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", global_drop)
def test_implicit_options_become_claude_automatic_caching(options: object, automatic: object):
result: Final = _resolved([{"role": "user", "content": "hi"}], None, {"prompt_cache_options": options, "top_p": 1})
assert drop_params_enabled(litellm_params) is expected
assert result.params == {"top_p": 1, "cache_control": automatic}
def test_implicit_options_leave_a_client_top_level_cache_control_alone():
hour: Final = {"type": "ephemeral", "ttl": "1h"}
result: Final = _resolved([], None, {"prompt_cache_options": {"mode": "implicit"}, "cache_control": hour})
assert result.params == {"cache_control": hour}
@pytest.mark.parametrize("options", [{"mode": "explicit"}, {}])
def test_options_without_implicit_mode_add_nothing_and_never_reach_claude(options: object):
assert _resolved([], None, {"prompt_cache_options": options}).params == {}
@pytest.mark.parametrize(
("tools", "expected"),
"options, accepts_automatic_caching, kept",
[
([_tool(prompt_cache_breakpoint=EXPLICIT)], True),
([{"type": "function", "function": {"name": "lookup", "prompt_cache_breakpoint": EXPLICIT}}], True),
((_tool(prompt_cache_breakpoint=EXPLICIT),), True),
([_tool(cache_control=EPHEMERAL), _tool()], False),
(None, False),
("not a list", False),
({"mode": "implicit"}, False, {}),
({"mode": "implicit", "ttl": "30m"}, True, {"cache_control": _EPHEMERAL}),
({"mode": "explicit", "ttl": "30m"}, True, {}),
({"mode": "always"}, True, {}),
("on", True, {}),
],
ids=["top-level", "under-function", "tuple", "cache-control-only", "none", "opaque"],
)
def test_tools_carry_prompt_cache_breakpoint_reads_both_tool_shapes(tools: object, expected: bool) -> None:
assert tools_carry_prompt_cache_breakpoint(tools) is expected
def test_options_claude_cannot_honor_are_reported_or_adjusted_under_drop_params(
options: object, accepts_automatic_caching: bool, kept: dict[str, object]
):
params: Final = {"prompt_cache_options": options}
problem: Final = _cache([], None, params, accepts_automatic_caching=accepts_automatic_caching)
dropped: Final = _resolved([], None, params, accepts_automatic_caching=accepts_automatic_caching, drop_params=True)
assert isinstance(problem, UnsupportedPromptCacheParam)
assert (problem.param, problem.value) == ("prompt_cache_options", options)
assert dropped.params == kept
def test_drop_params_keeps_the_latest_four_marks_in_the_order_claude_renders_them():
tools: Final = [{"type": "function", "function": {"name": "t"}, "prompt_cache_breakpoint": _EXPLICIT}]
messages: Final = [
{"role": "system", "content": [_part("s", cache_control=_EPHEMERAL)]},
{"role": "user", "content": [_part(str(index), prompt_cache_breakpoint=_EXPLICIT) for index in range(3)]},
]
result: Final = _resolved(messages, tools, {"prompt_cache_options": {"mode": "implicit"}}, drop_params=True)
assert list(result.tools or ()) == [{"type": "function", "function": {"name": "t"}}]
assert list(result.messages) == [
{"role": "system", "content": [_part("s")]},
{"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(3)]},
]
assert result.params == {"cache_control": _EPHEMERAL}
def test_without_drop_params_more_than_four_marks_are_sent_for_claude_to_answer():
messages: Final = [
{"role": "user", "content": [_part(str(index), prompt_cache_breakpoint=_EXPLICIT) for index in range(5)]}
]
result: Final = _resolved(messages)
assert list(result.messages) == [
{"role": "user", "content": [_part(str(index), cache_control=_EPHEMERAL) for index in range(5)]}
]

22
tests/unit/llms/azure_ai/claude/test_azure_anthropic_transformation.py
View file

@ -7,8 +7,6 @@ sys.path.insert(
from unittest.mock import MagicMock, patch
import json
import pytest
from litellm.llms.azure_ai.anthropic.transformation import AzureAnthropicConfig
@ -486,23 +484,3 @@ def test_chat_flagged_model_keeps_mid_conversation_system_role_in_place(local_mo
"role": "system",
"content": [{"type": "text", "text": "<system-reminder>Answer with exactly one word.</system-reminder>"}],
}
def test_prompt_cache_breakpoint_becomes_cache_control_on_the_same_block():
result = AzureAnthropicConfig().transform_request(
model="claude-opus-5-5",
messages=[
{
"role": "user",
"content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}}],
}
],
optional_params={},
litellm_params={},
headers={},
)
assert "prompt_cache_breakpoint" not in json.dumps(result)
assert result["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}}]}
]

136
tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py
View file

@ -873,49 +873,6 @@ async def test_bedrock_invoke_claude_async_completion_inlines_document_url_sourc
} in captured["body"]["messages"][0]["content"]
async def test_bedrock_invoke_claude_holds_extra_body_prompt_cache_options_to_drop_params(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
bodies: Final[list[dict[str, object]]] = [] # mutable-ok: the transport records each request
def handle(request: httpx.Request) -> httpx.Response:
bodies.append(json.loads(request.content))
return httpx.Response(
200,
json={
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "us.anthropic.claude-sonnet-5",
"content": [{"type": "text", "text": "hi"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
client: Final = AsyncHTTPHandler()
client.client = httpx.AsyncClient(transport=httpx.MockTransport(handle))
request: Final = {
"model": "bedrock/invoke/us.anthropic.claude-sonnet-5",
"messages": [{"role": "user", "content": "hi"}],
"extra_body": {"prompt_cache_options": {"mode": "explicit"}, "top_k": 3},
"aws_access_key_id": "AKIAEXAMPLE",
"aws_secret_access_key": "fake-secret",
"aws_region_name": "us-east-1",
"client": client,
}
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
await litellm.acompletion(**request)
assert bodies == []
await litellm.acompletion(**request, drop_params=True)
assert "prompt_cache_options" not in bodies[-1]
assert bodies[-1]["top_k"] == 3
@pytest.mark.parametrize(
"model, expected_betas",
[
@ -1394,97 +1351,6 @@ class TestBedrockAnthropicPromptCachingRegression:
has_cache_control = any(isinstance(c, dict) and "cache_control" in c for c in user_msg["content"])
assert has_cache_control, "cache_control should be present in invoke messages"
@staticmethod
def _claude_config(model_prefix: str):
from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
AmazonAnthropicClaudeConfig,
)
return AmazonConverseConfig() if "converse" in model_prefix else AmazonAnthropicClaudeConfig()
@pytest.mark.parametrize("model_prefix", ["bedrock/invoke/", "bedrock/converse/"])
def test_prompt_cache_breakpoint_lands_on_the_same_block_for_both_routes(self, model_prefix: str) -> None:
messages: Final = [
{
"role": "user",
"content": [
{
"type": "text",
"text": LARGE_DOCUMENT_FOR_CACHING,
"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"},
},
{"type": "text", "text": "What are the payment terms?"},
],
}
]
result: Final = self._claude_config(model_prefix).transform_request(
model="anthropic.claude-opus-5-5", messages=messages, optional_params={}, litellm_params={}, headers={}
)
assert "prompt_cache_breakpoint" not in json.dumps(result)
if "converse" in model_prefix:
assert result["messages"][0]["content"] == [
{"text": LARGE_DOCUMENT_FOR_CACHING},
{"cachePoint": {"type": "default"}},
{"text": "What are the payment terms?"},
]
else:
assert result["messages"][0]["content"] == [
{"type": "text", "text": LARGE_DOCUMENT_FOR_CACHING, "cache_control": {"type": "ephemeral"}},
{"type": "text", "text": "What are the payment terms?"},
]
@pytest.mark.parametrize("drop_params_source", ["litellm_params", "global"])
@pytest.mark.parametrize("model_prefix", ["bedrock/invoke/", "bedrock/converse/"])
def test_five_breakpoints_are_cut_to_four_only_under_drop_params_for_both_routes(
self, model_prefix: str, drop_params_source: str, monkeypatch: pytest.MonkeyPatch
) -> None:
import litellm
monkeypatch.setattr(litellm, "drop_params", False)
marked: Final = [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": {"mode": "explicit"}}
for index in range(5)
]
config: Final = self._claude_config(model_prefix)
with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 5"):
config.transform_request(
model="anthropic.claude-opus-5-5",
messages=[{"role": "user", "content": marked}],
optional_params={},
litellm_params={},
headers={},
)
monkeypatch.setattr(litellm, "drop_params", drop_params_source == "global")
result: Final = config.transform_request(
model="anthropic.claude-opus-5-5",
messages=[{"role": "user", "content": marked}],
optional_params={},
litellm_params={"drop_params": True} if drop_params_source == "litellm_params" else {},
headers={},
)
if "converse" in model_prefix:
assert result["messages"][0]["content"] == [
{"text": "part 0"},
*(
block
for index in range(1, 5)
for block in ({"text": f"part {index}"}, {"cachePoint": {"type": "default"}})
),
]
else:
assert result["messages"][0]["content"] == [
{"type": "text", "text": "part 0"},
*(
{"type": "text", "text": f"part {index}", "cache_control": {"type": "ephemeral"}}
for index in range(1, 5)
),
]
@pytest.mark.parametrize(
"model_prefix",
[
@ -1554,7 +1420,6 @@ class TestBedrockAnthropicPromptCachingRegression:
if "anthropic_beta" in additional_fields:
assert "prompt-caching-2024-07-31" not in additional_fields["anthropic_beta"]
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockAnthropic1MContextRegression:
"""
@ -1740,7 +1605,6 @@ class TestBedrockAnthropic1MContextRegression:
assert "context-1m-2025-08-07" in beta_headers
assert "computer-use-2024-10-22" in beta_headers
@pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown")
class TestBedrockAnthropicCombinedRegressions:
"""

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

@ -10145,326 +10145,3 @@ async def test_bedrock_user_file_element_becomes_document_block(sync_mode: bool)
assert content[0] == {"text": "Summarize this report"}
assert content[1]["document"]["source"] == {"bytes": base64.b64encode(pdf_bytes).decode()}
assert content[1]["document"]["format"] == "pdf"
_BREAKPOINT_MODEL: Final = "bedrock/anthropic.claude-opus-5-5"
_EXPLICIT: Final = {"mode": "explicit"}
_EXPLICIT_30M: Final = {"mode": "explicit", "ttl": "30m"}
_CACHE_POINT: Final = {"cachePoint": {"type": "default"}}
_LOOKUP_TOOL: Final = {
"type": "function",
"function": {
"name": "lookup",
"description": "lookup",
"parameters": {"type": "object", "properties": {}, "required": []},
},
}
_LOOKUP_TOOL_SPEC: Final = {
"toolSpec": {
"name": "lookup",
"description": "lookup",
"inputSchema": {"json": {"type": "object", "properties": {}, "required": []}},
}
}
def _breakpoint_conversation() -> list[dict[str, object]]:
return [
{"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": _EXPLICIT_30M}]},
{
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "lookup", "arguments": "{}"},
"prompt_cache_breakpoint": _EXPLICIT,
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [{"type": "text", "text": "sunny", "prompt_cache_breakpoint": _EXPLICIT}],
},
{"role": "user", "content": "done", "prompt_cache_breakpoint": _EXPLICIT},
]
def _expected_latest_four() -> dict[str, object]:
return {
"system": [{"text": "rules"}],
"toolConfig": {"tools": [_LOOKUP_TOOL_SPEC]},
"messages": [
{"role": "user", "content": [{"text": "hi"}, _CACHE_POINT]},
{
"role": "assistant",
"content": [{"toolUse": {"toolUseId": "call_1", "name": "lookup", "input": {}}}, _CACHE_POINT],
},
{
"role": "user",
"content": [
{"toolResult": {"content": [{"text": "sunny"}], "toolUseId": "call_1"}},
_CACHE_POINT,
{"text": "done"},
_CACHE_POINT,
],
},
],
}
async def _breakpoint_request(
sync_mode: bool, optional_params: dict[str, object], litellm_params: dict[str, object]
) -> dict[str, object]:
config: Final = AmazonConverseConfig()
if sync_mode:
return config._transform_request(
model=_BREAKPOINT_MODEL,
messages=_breakpoint_conversation(),
optional_params=optional_params,
litellm_params=litellm_params,
headers={},
)
return await config._async_transform_request(
model=_BREAKPOINT_MODEL,
messages=_breakpoint_conversation(),
optional_params=optional_params,
litellm_params=litellm_params,
headers={},
)
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_prompt_cache_breakpoints_become_cache_points_after_the_same_blocks(sync_mode: bool) -> None:
conversation: Final = _breakpoint_conversation()
config: Final = AmazonConverseConfig()
optional_params: Final[dict[str, object]] = {"tools": [{**_LOOKUP_TOOL, "prompt_cache_breakpoint": _EXPLICIT}]}
if sync_mode:
result = config._transform_request(
model=_BREAKPOINT_MODEL,
messages=conversation[2:],
optional_params=optional_params,
litellm_params={},
headers={},
)
else:
result = await config._async_transform_request(
model=_BREAKPOINT_MODEL,
messages=conversation[2:],
optional_params=optional_params,
litellm_params={},
headers={},
)
assert "prompt_cache_breakpoint" not in json.dumps(result)
assert result["toolConfig"]["tools"] == [_LOOKUP_TOOL_SPEC, _CACHE_POINT]
assert result["messages"] == _expected_latest_four()["messages"][1:]
@pytest.mark.parametrize("drop_params_source", ["litellm_params", "global"])
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_six_breakpoints_keep_the_latest_four_cache_points_under_drop_params(
sync_mode: bool, drop_params_source: str, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", drop_params_source == "global")
result: Final = await _breakpoint_request(
sync_mode,
{"tools": [{**_LOOKUP_TOOL, "prompt_cache_breakpoint": _EXPLICIT}]},
{"drop_params": True} if drop_params_source == "litellm_params" else {},
)
assert json.dumps(result).count("cachePoint") == 4
assert {key: result[key] for key in ("system", "toolConfig", "messages")} == _expected_latest_four()
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_six_breakpoints_are_a_bad_request_without_drop_params(
sync_mode: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 6"):
await _breakpoint_request(sync_mode, {"tools": [{**_LOOKUP_TOOL, "prompt_cache_breakpoint": _EXPLICIT}]}, {})
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_prompt_cache_options_are_rejected_unless_drop_params(
sync_mode: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
await _breakpoint_request(sync_mode, {"prompt_cache_options": {"mode": "explicit"}}, {})
dropped: Final = await _breakpoint_request(
sync_mode, {"prompt_cache_options": {"mode": "explicit"}}, {"drop_params": True}
)
assert "prompt_cache_options" not in json.dumps(dropped)
monkeypatch.setattr(litellm, "drop_params", True)
dropped_globally: Final = await _breakpoint_request(sync_mode, {"prompt_cache_options": {"mode": "explicit"}}, {})
assert "prompt_cache_options" not in json.dumps(dropped_globally)
_INFERENCE_PROFILE_ARN: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/abc123"
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_an_inference_profile_arn_gets_cache_points_and_the_prompt_cache_options_rule(
sync_mode: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
messages: Final = [
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": _EXPLICIT}]}
]
result: Final = await _breakpoint_converse_request(sync_mode, _INFERENCE_PROFILE_ARN, messages)
assert result["messages"] == [{"role": "user", "content": [{"text": "hi"}, _CACHE_POINT]}]
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
await _breakpoint_converse_request(
sync_mode, _INFERENCE_PROFILE_ARN, messages, optional_params={"prompt_cache_options": _EXPLICIT}
)
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_a_model_without_cache_points_gets_the_request_as_written(
sync_mode: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
monkeypatch.setitem(
litellm.model_cost,
_BREAKPOINT_MODEL,
{"litellm_provider": "bedrock", "mode": "chat", "supports_prompt_cache_breakpoint": False},
)
marked: Final = [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": _EXPLICIT} for index in range(5)
]
result: Final = await _breakpoint_converse_request(
sync_mode,
_BREAKPOINT_MODEL,
[{"role": "user", "content": marked}],
optional_params={"prompt_cache_options": {"mode": "implicit"}},
)
assert result["messages"] == [{"role": "user", "content": [{"text": f"part {index}"} for index in range(5)]}]
assert result["additionalModelRequestFields"] == {"prompt_cache_options": {"mode": "implicit"}}
def test_the_automatic_cache_control_takes_one_of_the_four_cache_points() -> None:
marked: Final = [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": _EXPLICIT} for index in range(4)
]
result: Final = AmazonConverseConfig()._transform_request(
model=_BREAKPOINT_MODEL,
messages=[{"role": "user", "content": marked}],
optional_params={"cache_control": {"type": "ephemeral"}},
litellm_params={"drop_params": True},
headers={},
)
assert result["messages"] == [
{
"role": "user",
"content": [
{"text": "part 0"},
*(block for index in range(1, 4) for block in ({"text": f"part {index}"}, _CACHE_POINT)),
],
}
]
def test_five_native_cache_control_marks_are_forwarded_as_written_past_the_cap() -> None:
marked: Final = [
{"type": "text", "text": f"part {index}", "cache_control": {"type": "ephemeral"}} for index in range(5)
]
result: Final = AmazonConverseConfig()._transform_request(
model=_BREAKPOINT_MODEL,
messages=[{"role": "user", "content": marked}],
optional_params={},
litellm_params={"drop_params": True},
headers={},
)
assert result["messages"] == [
{
"role": "user",
"content": [block for index in range(5) for block in ({"text": f"part {index}"}, _CACHE_POINT)],
}
]
async def _breakpoint_converse_request(
sync_mode: bool, model: str, messages: list[dict[str, object]], **params: dict[str, object]
) -> dict[str, object]:
config: Final = AmazonConverseConfig()
optional_params: Final = params.get("optional_params", {})
litellm_params: Final = params.get("litellm_params", {})
if sync_mode:
return config._transform_request(
model=model, messages=messages, optional_params=optional_params, litellm_params=litellm_params, headers={}
)
return await config._async_transform_request(
model=model, messages=messages, optional_params=optional_params, litellm_params=litellm_params, headers={}
)
@pytest.mark.parametrize("drop_params", [False, True], ids=["strict", "drop-params"])
@pytest.mark.parametrize("sync_mode", [True, False], ids=["sync", "async"])
@pytest.mark.asyncio
async def test_a_tool_result_with_two_marked_parts_is_one_cache_point_within_the_cap(
sync_mode: bool, drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
messages: Final = [
{"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": _EXPLICIT}]},
{
"role": "assistant",
"content": "",
"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": [
{"type": "text", "text": "sunny", "prompt_cache_breakpoint": _EXPLICIT},
{"type": "text", "text": "warm", "prompt_cache_breakpoint": _EXPLICIT},
],
},
{"role": "user", "content": "done", "prompt_cache_breakpoint": _EXPLICIT},
]
result: Final = await _breakpoint_converse_request(
sync_mode,
_BREAKPOINT_MODEL,
messages,
optional_params={"tools": [_LOOKUP_TOOL]},
litellm_params={"drop_params": drop_params},
)
assert result["system"] == [{"text": "rules"}, _CACHE_POINT]
assert result["messages"] == [
{"role": "user", "content": [{"text": "hi"}, _CACHE_POINT]},
{"role": "assistant", "content": [{"toolUse": {"toolUseId": "call_1", "name": "lookup", "input": {}}}]},
{
"role": "user",
"content": [
{"toolResult": {"content": [{"text": "sunny"}, {"text": "warm"}], "toolUseId": "call_1"}},
_CACHE_POINT,
{"text": "done"},
_CACHE_POINT,
],
},
]

98
tests/unit/llms/bedrock/test_claude_platform_provider.py
View file

@ -718,104 +718,6 @@ def test_chat_completion_claude_platform_override_kwarg_is_honoured_and_not_sent
assert requests[0]["body"] == {**EXPECTED_CHAT_BODY, "context_management": context_management}
EXPLICIT_BREAKPOINT: Final = {"mode": "explicit"}
EPHEMERAL: Final = {"type": "ephemeral"}
def test_chat_completion_claude_platform_sends_prompt_cache_breakpoints_as_cache_control(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import litellm
monkeypatch.setattr(litellm, "drop_params", False)
requests: Final[list[dict]] = []
litellm.completion(
model="bedrock/claude_platform/claude-sonnet-4-6",
messages=[
{
"role": "system",
"content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": EXPLICIT_BREAKPOINT}],
},
{"role": "user", "content": "hello", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}},
],
max_tokens=10,
workspace_id="wrkspc_test",
client=_sync_gateway_client(requests),
**API_KEY_KWARGS,
)
assert requests[0]["body"]["system"] == [{"type": "text", "text": "rules", "cache_control": EPHEMERAL}]
assert requests[0]["body"]["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "hello", "cache_control": EPHEMERAL}]}
]
def test_chat_completion_claude_platform_holds_breakpoints_to_four_unless_drop_params(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import litellm
monkeypatch.setattr(litellm, "drop_params", False)
requests: Final[list[dict]] = []
request: Final = {
"model": "bedrock/claude_platform/claude-sonnet-4-6",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": f"part {index}", "prompt_cache_breakpoint": EXPLICIT_BREAKPOINT}
for index in range(5)
],
}
],
"max_tokens": 10,
"workspace_id": "wrkspc_test",
"client": _sync_gateway_client(requests),
**API_KEY_KWARGS,
}
with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 5"):
litellm.completion(**request)
assert requests == []
litellm.completion(**request, drop_params=True)
assert requests[0]["body"]["messages"][0]["content"] == [
{"type": "text", "text": "part 0"},
*({"type": "text", "text": f"part {index}", "cache_control": EPHEMERAL} for index in range(1, 5)),
]
@pytest.mark.parametrize("source", ["top-level", "extra_body"])
def test_chat_completion_claude_platform_rejects_prompt_cache_options_unless_drop_params(
source: str, monkeypatch: pytest.MonkeyPatch
) -> None:
import litellm
monkeypatch.setattr(litellm, "drop_params", False)
requests: Final[list[dict]] = []
options: Final = {"prompt_cache_options": EXPLICIT_BREAKPOINT}
request: Final = {
"model": "bedrock/claude_platform/claude-sonnet-4-6",
"messages": [{"role": "user", "content": "hello"}],
"max_tokens": 10,
"workspace_id": "wrkspc_test",
"client": _sync_gateway_client(requests),
**(options if source == "top-level" else {"extra_body": options}),
**API_KEY_KWARGS,
}
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
litellm.completion(**request)
assert requests == []
response: Final = litellm.completion(**request, drop_params=True)
assert response.choices[0].message.content == "ok"
assert requests[0]["body"] == EXPECTED_CHAT_BODY
def test_fake_claude_platform_gateway_rejects_leaked_internal_fields():
leaked = json.dumps({**EXPECTED_NATIVE_BODY, "workspace_id": "wrkspc_test", "aws_region_name": "us-west-2"})
response = _fake_claude_platform_gateway(

97
tests/unit/llms/databricks/chat/test_databricks_chat_transformation.py
View file

@ -2255,100 +2255,3 @@ def test_transform_response_carries_served_service_tier(served_tier: str | None)
)
assert result.to_dict().get("service_tier") == served_tier
_EXPLICIT: Final = {"mode": "explicit"}
_EPHEMERAL: Final = {"type": "ephemeral"}
_LOOKUP_FUNCTION: Final = {"name": "lookup", "parameters": {"type": "object", "properties": {}}}
def _databricks_claude_route(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch) -> respx.Route:
monkeypatch.setenv("DATABRICKS_API_BASE", DATABRICKS_API_BASE)
monkeypatch.setenv("DATABRICKS_API_KEY", DATABRICKS_API_KEY)
monkeypatch.setattr(litellm, "drop_params", False)
return respx_mock.post(DATABRICKS_CHAT_COMPLETIONS_URL).mock(
return_value=httpx.Response(
200, json=_databricks_anthropic_cache_response(cache_read_input_tokens=0, cache_creation_input_tokens=0)
)
)
def test_prompt_cache_breakpoints_reach_databricks_claude_as_cache_control(
monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
) -> None:
route: Final = _databricks_claude_route(respx_mock, monkeypatch)
litellm.completion(
model="databricks/databricks-claude-haiku-4-5",
messages=[
{"role": "system", "content": [{"type": "text", "text": "rules", "prompt_cache_breakpoint": _EXPLICIT}]},
{"role": "user", "content": "hi", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}},
],
)
assert json.loads(route.calls.last.request.content)["messages"] == [
{"role": "system", "content": [{"type": "text", "text": "rules", "cache_control": _EPHEMERAL}]},
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": _EPHEMERAL}]},
]
def test_a_tool_breakpoint_holds_databricks_claude_to_four_marks_under_drop_params(
monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
) -> None:
route: Final = _databricks_claude_route(respx_mock, monkeypatch)
parts: Final = [{"type": "text", "text": f"part {index}", "cache_control": _EPHEMERAL} for index in range(5)]
request: Final = {
"model": "databricks/databricks-claude-haiku-4-5",
"messages": [{"role": "user", "content": parts}],
"tools": [{"type": "function", "function": _LOOKUP_FUNCTION, "prompt_cache_breakpoint": _EXPLICIT}],
}
with pytest.raises(litellm.BadRequestError, match=r"Claude accepts at most 4 cache breakpoints, found 5"):
litellm.completion(**request)
assert route.call_count == 0
litellm.completion(**request, drop_params=True)
assert json.loads(route.calls.last.request.content)["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "part 0"}, *parts[1:]]}
]
def _prompt_cache_options_param(source: str) -> dict[str, object]:
options: Final = {"prompt_cache_options": {"mode": "explicit"}}
return options if source == "top-level" else {"extra_body": options}
@pytest.mark.parametrize("source", ["top-level", "extra_body"])
def test_prompt_cache_options_toward_databricks_claude_are_rejected_unless_drop_params(
source: str, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
) -> None:
route: Final = _databricks_claude_route(respx_mock, monkeypatch)
request: Final = {
"model": "databricks/databricks-claude-haiku-4-5",
"messages": [{"role": "user", "content": "hi"}],
**_prompt_cache_options_param(source),
}
with pytest.raises(litellm.UnsupportedParamsError, match="drop_params"):
litellm.completion(**request)
assert route.call_count == 0
litellm.completion(**request, drop_params=True)
assert "prompt_cache_options" not in json.loads(route.calls.last.request.content)
@pytest.mark.parametrize("source", ["top-level", "extra_body"])
def test_prompt_cache_options_toward_a_non_claude_databricks_model_are_forwarded_as_before(
source: str, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter
) -> None:
route: Final = _databricks_claude_route(respx_mock, monkeypatch)
litellm.completion(
model="databricks/databricks-meta-llama-3-3-70b-instruct",
messages=[{"role": "user", "content": "hi"}],
**_prompt_cache_options_param(source),
)
assert json.loads(route.calls.last.request.content)["prompt_cache_options"] == {"mode": "explicit"}

61
tests/unit/llms/vertex_ai/vertex_ai_partner_models/anthropic/test_vertex_ai_partner_models_anthropic_transformation.py
View file

@ -1090,64 +1090,3 @@ def test_chat_flagged_model_replays_a_byte_identical_prefix_around_a_mid_convers
_assert_prefix_stable(requests)
assert [m["role"] for m in requests[1]["messages"]] == ["user", "assistant", "user", "system"]
assert [m["role"] for m in requests[2]["messages"]] == ["user", "assistant", "user", "system", "assistant", "user"]
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_vertex_ai_anthropic_prompt_cache_breakpoint_becomes_cache_control_on_the_same_block(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sync_mode: bool,
) -> None:
from litellm.main import vertex_partner_models_chat_completion
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setitem(
vertex_partner_models_chat_completion._credentials_project_mapping,
("test-vertex-credentials", "test-project"),
(Credentials(token="test-vertex-token"), "test-project"),
)
route: Final = respx_mock.post(
"https://us-east5-aiplatform.googleapis.com/v1/projects/test-project/locations/us-east5/"
"publishers/anthropic/models/claude-opus-5-5:rawPredict"
).mock(
return_value=httpx.Response(
200,
json={
"id": "msg-breakpoint",
"type": "message",
"role": "assistant",
"model": "claude-opus-5-5",
"content": [{"type": "text", "text": "cached"}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 4, "output_tokens": 1},
},
)
)
call_params: Final = {
"model": "vertex_ai/claude-opus-5-5",
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "Cache this request", "prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"}}
],
}
],
"vertex_project": "test-project",
"vertex_location": "us-east5",
"vertex_credentials": "test-vertex-credentials",
"max_retries": 0,
}
if sync_mode:
litellm.completion(**call_params)
else:
await litellm.acompletion(**call_params)
assert route.call_count == 1
request: Final = route.calls[0].request
assert "prompt_cache_breakpoint" not in request.content.decode()
assert json.loads(request.content)["messages"] == [
{"role": "user", "content": [{"type": "text", "text": "Cache this request", "cache_control": {"type": "ephemeral"}}]}
]