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

fix(cache_control)!: keep one deployment's prompt-cache opt-in from reaching siblings on the same model (#49685)

Browse source 
* fix(cache_control): read the routed deployment's own prompt-cache flag, not a sibling's shared cost-map entry

Two deployments that share one model string share one key in the live
litellm.model_cost map, and router registration writes each deployment's
supports_prompt_cache_breakpoint into that key. The cache control hook read
the live map, so a flagless deployment picked up a sibling's opt-in and sent
prompt_cache_options and prompt_cache_breakpoint upstream, which the provider
rejects with a 400

When a call carries the routed deployment's model_info, the prompt cache
checks now use its declared supports_prompt_cache_breakpoint, and fall back
to the loaded catalog instead of the live map. This covers the hook on
completion, acompletion, Responses and the /v1/messages route, the azure_ai
chat transform, the OpenAI Responses transform under drop_params, and the
/v1/messages to Responses adapter

* fix(prompt caching): read each deployment's own breakpoint flag in the openai and bedrock readers

A deployment's own supports_prompt_cache_breakpoint now picks OpenAI's
prompt_cache_breakpoint only where the request goes out OpenAI-shaped on a
provider other than openai, such as azure_ai OpenAI models and bedrock_mantle.
Bedrock on every route and providers that build Anthropic messages keep
cache_control, which Bedrock Converse turns into cachePoint

The openai chat reader now passes the routed model_info and checks base_model.
The Bedrock cache point reader takes the deployment's flag first and, for a
routed call with no flag, reads the catalog as loaded instead of the live map
that sibling deployments write into. model_info is threaded from the converse
transform, the factory message and tool builders, and the responses bridge

* fix(prompt caching): keep gateways on cache_control and read register_model opt-ins on routed calls

A declared supports_prompt_cache_breakpoint picks OpenAI's breakpoint only on providers hosting OpenAI's API or for a model the cost map lists under the provider, so OpenRouter, LiteLLM proxy, and vLLM deployments keep cache_control as before. Routed calls read the loaded catalog plus the user's own register_model entries, never router registrations.

* fix(prompt caching): honor a deployment's own supports_prompt_caching on Bedrock and its declared litellm_provider on gateways

* fix(prompt caching): keep wildcard cache flags off matched models, read deployment flags for auto caching and base_model aliases

* test(anthropic): cover auto caching on an unmapped Claude deployment

* fix(cache_control): keep the merge base's reads where the per-deployment flag changed them

A register_model entry under a provider-prefixed name reaches routed calls again, since the
overlay now resolves the bare and prefixed names the way runtime registrations are stored.

The hook and the openai chat reader no longer judge base_model, as at the merge base, so a
client base_model sent under allow_client_pricing_override cannot pick OpenAI's breakpoint
and get a 400.

A Bedrock deployment flag that is not a real true reads as false again, as at the merge base.

An azure_ai wildcard's supports_prompt_cache_breakpoint speaks for each model it matches again,
as at the merge base. Other wildcards still leave both cache flags to the catalog.

The s10/s11 Bedrock wire test asserted that a flagless deployment inherits its sibling's flag
on a shared ARN, which is the leak this PR removes. It now asserts that each deployment keeps
its own flag and that a flagless unmapped ARN keeps its cache points after the delete.

* fix(cache_control): let the later register_model entry win between a bare and a provider-prefixed name

Routed calls read register_model entries through an overlay that merged the bare
name after the provider-prefixed one whatever order they were registered in, so a
later opt-out under one name lost to an earlier opt-in under the other. Each
registration now moves its key to the end of the kept registrations, and the
overlay merges both names oldest first, which matches what SDK calls read from
the live model map.

Also lists the wildcard cache flag helper in the router coverage script, since
the hook's wildcard tests drive it rather than calling it by name.

* fix(cache_control): keep register_model entries visible while a cost map reload replays them

The previous commit moved a re-registered name to the end of the saved
registrations by removing and re-adding it, and the overlay walked that dict to
order a bare and a provider-prefixed name. A reload replays every saved entry,
so a request on another thread could briefly miss an entry or fail on a dict
that changed size mid-walk. Entries are written in place again, a per-name
sequence records which registration came last, and both the overlay and the
reload replay order by it, so the later registration still wins after a reload.

* fix(cache_control): read a string or numeric supports_prompt_caching as the bool it spells and find register_model entries under any route prefix

A deployment flag stored as "true", 1, "false" or 0 (POST /model/new and quoted YAML keep strings) now means that bool on the Bedrock cachePoint reader and Anthropic auto caching, and any other value falls to the catalog. register_model records which names wrote to each catalog row, so an opt-out under bedrock/ or bedrock/converse/ reaches routed calls like one under anthropic/.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
tessellark 2026-11-02 23:55:51 -08:00 • committed by GitHub
parent ecb8ef5ca8
commit 5b546259d5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
26 changed files with 1511 additions and 116 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -34,7 +34,7 @@ from litellm.completion_extras.litellm_responses_transformation.history import (
replay_items,
requested_phase,
)
from litellm.integrations.anthropic_cache_control_hook import supports_openai_prompt_cache_breakpoint
from litellm.integrations.anthropic_cache_control_hook import routed_model_info, supports_openai_prompt_cache_breakpoint
from litellm.litellm_core_utils.allowed_tools_tool_choice import (
parse_chat_allowed_tools_tool_choice,
responses_allowed_tools_tool_choice,
@ -1166,10 +1166,10 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
optional_params = self._extract_extra_body_params(optional_params)
provider: Final = _served_provider(litellm_params)
base_model: Final = litellm_params.get("base_model")
supports_prompt_cache_breakpoint: Final = supports_openai_prompt_cache_breakpoint(model, provider) or (
isinstance(base_model, str)
and bool(base_model)
and supports_openai_prompt_cache_breakpoint(base_model, provider)
model_info: Final = routed_model_info(litellm_params.get("model_info"))
named_models: Final = (model, base_model) if isinstance(base_model, str) and base_model else (model,)
supports_prompt_cache_breakpoint: Final = any(
supports_openai_prompt_cache_breakpoint(name, provider, model_info) for name in named_models
)
converted_input_items, converted_instructions = self.convert_chat_completion_messages_to_responses_api(
messages,

207
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -14,6 +14,7 @@ import itertools
import os
import re
from collections.abc import Callable, Container, Iterable, Iterator, Mapping, Sequence
from functools import lru_cache
from typing import TYPE_CHECKING, Any, Final, TypeAlias, TypeVar, cast
from urllib.parse import urlparse
@ -23,7 +24,6 @@ from litellm._logging import verbose_logger
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.custom_prompt_management import CustomPromptManagement
from litellm.integrations.prompt_management_base import PromptManagementClient
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.prompt_templates.common_utils import (
cache_control_as_prompt_cache_breakpoint,
with_prompt_cache_breakpoint,
@ -85,8 +85,40 @@ _OBJECT_MAPPING_ADAPTER: Final = TypeAdapter(dict[object, object])
_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
_CONTENT_PART_ADAPTER: Final = TypeAdapter(dict[str, object])
_ATTRIBUTE_VALUE: Final = TypeAdapter(object)
_MODEL_INFO_ADAPTER: Final = TypeAdapter(Mapping[str, object])
AllToolParamValues = ChatCompletionToolParam | AllAnthropicToolsValues
CostMap: TypeAlias = Mapping[str, Mapping[str, object]]
def routed_model_info(value: object) -> Mapping[str, object] | None:
try:
return _MODEL_INFO_ADAPTER.validate_python(value)
except ValidationError:
return None
def _live_cost_map_unless(model_cost: CostMap | None) -> CostMap:
import litellm
if model_cost is not None:
return model_cost
return cast(CostMap, litellm.model_cost) # cast-ok: the live map holds catalog entries
def declared_prompt_cache_breakpoint(model_info: Mapping[str, object] | None) -> bool | None:
declared: Final = None if model_info is None else model_info.get("supports_prompt_cache_breakpoint")
return declared if isinstance(declared, bool) else None
def routed_cost_map() -> CostMap:
from litellm.utils import catalog_with_runtime_registrations
return catalog_with_runtime_registrations()
def _catalog_unless_unrouted(model_info: Mapping[str, object] | None) -> CostMap | None:
return None if model_info is None else routed_cost_map()
def _validated_object_mapping(value: object) -> dict[object, object] | None:
@ -314,19 +346,27 @@ def _served_provider_for_model(model: str, api_base: object = None) -> str | Non
return provider
def supports_openai_prompt_cache_breakpoint(model: str, custom_llm_provider: str | None = None) -> bool:
return openai_prompt_cache_breakpoint_support(model, custom_llm_provider) is True
def supports_openai_prompt_cache_breakpoint(
model: str, custom_llm_provider: str | None = None, model_info: Mapping[str, object] | None = None
) -> bool:
return openai_prompt_cache_breakpoint_support(model, custom_llm_provider, model_info) is True
def openai_prompt_cache_breakpoint_support(model: str, custom_llm_provider: str | None = None) -> bool | None:
def openai_prompt_cache_breakpoint_support(
model: str, custom_llm_provider: str | None = None, model_info: Mapping[str, object] | None = None
) -> bool | None:
declared: Final = declared_prompt_cache_breakpoint(model_info)
if declared is not None:
return declared
model_cost: Final = _catalog_unless_unrouted(model_info)
hosted_flag: Final = (
None
if custom_llm_provider is None
else _hosted_openai_dialect_flag(model, custom_llm_provider, _served_provider_for_model)
else _hosted_openai_dialect_flag(model, custom_llm_provider, _served_provider_for_model, model_cost)
)
if hosted_flag is not None:
return hosted_flag
model_map_flag: Final = _model_map_prompt_cache_breakpoint_flag(model)
model_map_flag: Final = _model_map_prompt_cache_breakpoint_flag(model, model_cost)
if model_map_flag is not None:
return model_map_flag
fine_tuned_base: Final = _FINE_TUNED_BASE_PATTERN.match(model.rsplit("/", 1)[-1])
@ -347,13 +387,18 @@ class _DeclaredBaseModels(BaseModel):
return tuple(name for name in candidates if isinstance(name, str) and name)
def deployment_model_info(litellm_params: object) -> Mapping[str, object] | None:
return routed_model_info(_DeclaredBaseModels.model_validate(litellm_params).model_info)
def deployment_prompt_cache_breakpoint_support(
model: str, custom_llm_provider: str, litellm_params: object
) -> bool | None:
"""``openai_prompt_cache_breakpoint_support`` for a deployment, read from its declared base model first."""
declared: Final = _DeclaredBaseModels.model_validate(litellm_params)
model_info: Final = routed_model_info(declared.model_info)
supports: Final = (
openai_prompt_cache_breakpoint_support(name, custom_llm_provider)
openai_prompt_cache_breakpoint_support(name, custom_llm_provider, model_info)
for name in dict.fromkeys((*declared.names(), model))
)
return next((support for support in supports if support is not None), None)
@ -370,11 +415,10 @@ def _named_model_breakpoint_support(model: str) -> bool | None:
return version >= OPENAI_PROMPT_CACHE_BREAKPOINT_MIN_GPT_VERSION
def _model_map_prompt_cache_breakpoint_flag(model: str) -> bool | None:
import litellm
entries: Final = (litellm.model_cost.get(key) for key in (model, model.rsplit("/", 1)[-1]))
flags: Final = (entry.get("supports_prompt_cache_breakpoint") for entry in entries if isinstance(entry, dict))
def _model_map_prompt_cache_breakpoint_flag(model: str, model_cost: CostMap | None = None) -> bool | None:
cost_map: Final = _live_cost_map_unless(model_cost)
entries: Final = (cost_map.get(key) for key in (model, model.rsplit("/", 1)[-1]))
flags: Final = (entry.get("supports_prompt_cache_breakpoint") for entry in entries if isinstance(entry, Mapping))
return next((flag is True for flag in flags if flag is not None), None)
@ -390,7 +434,7 @@ def _hosted_openai_dialect_flag(
model: str,
custom_llm_provider: str | None,
resolve_provider: Callable[[str], str | None],
model_cost: Mapping[str, Mapping[str, object]] | None = None,
model_cost: CostMap | None = None,
) -> bool | None:
"""
Explicit opt-in for an OpenAI-shaped deployment served by another provider.
@ -407,13 +451,7 @@ def _hosted_openai_dialect_flag(
row of the same name, so an exact entry only speaks for a deployment when it is
keyed for that deployment's provider.
"""
import litellm
cost_map: Final = (
cast(Mapping[str, Mapping[str, object]], litellm.model_cost) # cast-ok: the live map holds catalog entries
if model_cost is None
else model_cost
)
cost_map: Final = _live_cost_map_unless(model_cost)
exact_entry: Final = cost_map.get(model)
exact_provider: Final = exact_entry.get("litellm_provider") if isinstance(exact_entry, Mapping) else None
if custom_llm_provider is None and (exact_provider == "openai" or ("/" not in model and exact_provider is None)):
@ -423,10 +461,16 @@ def _hosted_openai_dialect_flag(
return None
if isinstance(exact_entry, Mapping) and exact_provider == provider:
return _hosted_entry_flag(exact_entry, lambda: provider)
entries: Final = (cost_map.get(candidate) for candidate in _provider_cost_map_keys(model, provider))
flags: Final = (_hosted_entry_flag(entry, lambda: provider) for entry in entries if isinstance(entry, Mapping))
return next((flag for flag in flags if flag is not None), None)
def _provider_cost_map_keys(model: str, provider: str) -> tuple[str, ...]:
from litellm.utils import get_potential_model_names
names: Final = get_potential_model_names(model, provider)
candidates: Final = (
return (
names["combined_model_name"],
names["region_free_combined_model_name"],
names["split_model"],
@ -434,9 +478,60 @@ def _hosted_openai_dialect_flag(
names["stripped_model_name"],
names["provider_prefixed_model_name"],
)
entries: Final = (cost_map.get(candidate) for candidate in candidates)
flags: Final = (_hosted_entry_flag(entry, lambda: provider) for entry in entries if isinstance(entry, Mapping))
return next((flag for flag in flags if flag is not None), None)
def _cost_map_lists_model_for_provider(model: str, provider: str, model_cost: CostMap) -> bool:
entries: Final = (model_cost.get(key) for key in (model, *_provider_cost_map_keys(model, provider)))
return any(isinstance(entry, Mapping) and entry.get("litellm_provider") == provider for entry in entries)
_PROVIDERS_READING_THE_FLAG_AS_CACHE_POINT: Final = frozenset(("bedrock", "bedrock_converse"))
_PROVIDERS_HOSTING_OPENAIS_API: Final = frozenset(("azure", "azure_ai", "bedrock_mantle"))
def _declared_hosted_flag(
declared: bool, model: str, provider: str | None, model_cost: CostMap, deployment_provider: object
) -> bool | None:
"""
A deployment's own ``supports_prompt_cache_breakpoint`` picks OpenAI's breakpoint only where the
request goes out OpenAI-shaped. Bedrock reads the same flag on every route as whether Converse takes
a ``cachePoint``, and a provider building Anthropic messages takes ``cache_control``, so both keep
the Anthropic mark. A gateway or self-hosted server (OpenRouter, a LiteLLM proxy, vLLM) may pass
the request on to Anthropic, so there the flag only picks OpenAI's breakpoint for a model the cost
map, or the deployment's own ``litellm_provider``, lists under that provider.
"""
if provider is None or provider == "openai":
return None
return (
declared
and provider not in _PROVIDERS_READING_THE_FLAG_AS_CACHE_POINT
and (
provider in _PROVIDERS_HOSTING_OPENAIS_API
or deployment_provider == provider
or _cost_map_lists_model_for_provider(model, provider, model_cost)
)
and _sends_openai_shaped_chat(model, provider)
)
@lru_cache(maxsize=512)
def _sends_openai_shaped_chat(model: str, custom_llm_provider: str) -> bool:
from litellm.exceptions import BadRequestError
from litellm.litellm_core_utils.get_llm_provider_logic import resolve_model_and_provider_for_metadata
from litellm.llms.azure.chat.gpt_transformation import AzureOpenAIConfig
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.llms.openai.openai import OpenAIConfig
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
try:
served_model, provider = resolve_model_and_provider_for_metadata(model, custom_llm_provider)
chat_config: Final = ProviderConfigManager.get_provider_chat_config(
model=served_model, provider=LlmProviders(provider)
)
except (BadRequestError, ValueError):
return False
return isinstance(chat_config, (OpenAIGPTConfig, OpenAIConfig, AzureOpenAIConfig))
def targets_openai_api(api_base: object) -> bool:
@ -677,7 +772,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
custom_llm_provider: Final = _custom_llm_provider(non_default_params, injection_points[0])
request_params: Final = cast(Mapping[str, object], non_default_params) # cast-ok: base hook's bare dict
custom_llm_provider: Final = _custom_llm_provider(request_params, injection_points[0])
stamped_dialect: Final = injection_points[0].get("_litellm_openai_dialect")
openai_dialect: Final = (
stamped_dialect
@ -685,8 +781,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
else AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model,
custom_llm_provider,
non_default_params.get("api_base") or non_default_params.get("base_url"),
non_default_params.get("prompt_cache_options"),
request_params.get("api_base") or request_params.get("base_url"),
request_params.get("prompt_cache_options"),
routed_model_info(request_params.get("model_info")),
)
)
# A provisional message list defers every role-targeted point to the pass holding
@ -757,32 +854,35 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider: str | None,
api_base: object = None,
prompt_cache_options: object = None,
model_info: Mapping[str, object] | None = None,
) -> bool:
if model is None:
return False
hosted_flag: Final = _hosted_openai_dialect_flag(
model, custom_llm_provider, lambda name: AnthropicCacheControlHook._resolve_provider(name, api_base)
declared: Final = declared_prompt_cache_breakpoint(model_info)
hosted_flag: Final = (
_hosted_openai_dialect_flag(
model,
custom_llm_provider,
lambda name: AnthropicCacheControlHook._resolve_provider(name, api_base),
_catalog_unless_unrouted(model_info),
)
if declared is None
else _declared_hosted_flag(
declared,
model,
custom_llm_provider or AnthropicCacheControlHook._resolve_provider(model, api_base),
routed_cost_map(),
None if model_info is None else model_info.get("litellm_provider"),
)
)
if hosted_flag is not None:
return hosted_flag
if not supports_openai_prompt_cache_breakpoint(model):
if not supports_openai_prompt_cache_breakpoint(model, model_info=model_info):
return False
if (custom_llm_provider or AnthropicCacheControlHook._resolve_provider(model)) != "openai":
return False
return prompt_cache_options is not None or targets_openai_api(api_base)
@staticmethod
def catalog_targets_openai_prompt_cache_breakpoint(model: str, custom_llm_provider: str) -> bool:
return (
_hosted_openai_dialect_flag(
model,
custom_llm_provider,
AnthropicCacheControlHook._resolve_provider,
GetModelCostMap.loaded_model_cost_map(),
)
is True
)
@staticmethod
def _resolve_provider(model: str, api_base: object = None) -> str | None:
return _served_provider_for_model(model, api_base)
@ -1181,6 +1281,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider: str | None,
api_base: object,
prompt_cache_options: object,
model_info: Mapping[str, object] | None = None,
) -> Sequence[Mapping[str, object]]:
"""Carry onto the points what the prompt-management hook never receives.
@ -1192,7 +1293,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
across requests.
"""
with_dialect: Final = AnthropicCacheControlHook._stamped_with_dialect(
points, model, custom_llm_provider, api_base, prompt_cache_options
points, model, custom_llm_provider, api_base, prompt_cache_options, model_info
)
stamped_earlier: Final = any(EXTERNAL_BREAKPOINTS_STAMP in point for point in with_dialect)
if external_breakpoints == 0 and not stamped_earlier:
@ -1206,14 +1307,17 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider: str | None,
api_base: object,
prompt_cache_options: object,
model_info: Mapping[str, object] | None = None,
) -> Sequence[Mapping[str, object]]:
if not AnthropicCacheControlHook._may_target_openai_prompt_cache_breakpoint(model, custom_llm_provider):
if model_info is None and not AnthropicCacheControlHook._may_target_openai_prompt_cache_breakpoint(
model, custom_llm_provider
):
return points
return AnthropicCacheControlHook._stamped(
points,
"_litellm_openai_dialect",
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, custom_llm_provider, api_base, prompt_cache_options
model, custom_llm_provider, api_base, prompt_cache_options, model_info
),
)
@ -1272,6 +1376,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control: object = None,
request_kwargs: object = None,
on_messages_route: bool = False,
model_info: Mapping[str, object] | None = None,
) -> list[CacheControlInjectionPoint]:
"""Default breakpoints when ``litellm.enable_anthropic_prompt_caching`` is on.
@ -1289,7 +1394,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if litellm.enable_anthropic_prompt_caching is not True and enable_prompt_caching is not True:
return []
if not supports_anthropic_cache_control(model, custom_llm_provider):
if not supports_anthropic_cache_control(model, custom_llm_provider, model_info):
return []
if AnthropicCacheControlHook._request_has_cache_control(
@ -1367,6 +1472,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
tools: list | None = None,
enable_prompt_caching: bool | None = None,
api_base: object = None,
model_info: Mapping[str, object] | None = None,
) -> None:
"""For /chat/completions: resolve the injection points the request should carry.
@ -1393,6 +1499,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider,
api_base,
non_default_params.get("prompt_cache_options"),
model_info,
)
non_default_params["cache_control_injection_points"] = (
AnthropicCacheControlHook._stamped(stamped, CUSTOM_LLM_PROVIDER_STAMP, custom_llm_provider)
@ -1409,6 +1516,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
enable_prompt_caching=enable_prompt_caching,
cache_control=non_default_params.get("cache_control"),
request_kwargs=non_default_params,
model_info=model_info,
)
if points:
non_default_params["cache_control_injection_points"] = points
@ -1527,6 +1635,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control=cache_control,
request_kwargs=kwargs,
on_messages_route=True,
model_info=routed_model_info(kwargs.get("model_info")),
)
if model is not None
else ()
@ -1535,7 +1644,11 @@ class AnthropicCacheControlHook(CustomPromptManagement):
return messages, system
openai_dialect: Final = AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, custom_llm_provider, api_base, kwargs.get("prompt_cache_options")
model,
custom_llm_provider,
api_base,
kwargs.get("prompt_cache_options"),
routed_model_info(kwargs.get("model_info")),
)
breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system)
messages, system, remaining = AnthropicCacheControlHook.apply_to_anthropic_messages_request(

10
litellm/litellm_core_utils/core_helpers.py
View file

@ -97,7 +97,7 @@ def safe_divide_seconds(seconds: float, denominator: float, default: float | Non
return float(seconds / denominator)
_DROP_PARAMS_BOOL: Final = TypeAdapter(bool)
_LAX_BOOL: Final = TypeAdapter(bool)
def as_str_mapping(value: object) -> Mapping[str, object] | None:
@ -160,15 +160,19 @@ def attribute_items(source: object, name: str) -> Iterable[object]:
return iter_items(value) if value else ()
def normalize_drop_params(value: object) -> bool | None:
def lax_bool(value: object) -> bool | None:
if value is None or isinstance(value, bool):
return value
try:
return _DROP_PARAMS_BOOL.validate_python(value.strip() if isinstance(value, str) else value)
return _LAX_BOOL.validate_python(value.strip() if isinstance(value, str) else value)
except ValidationError:
return None
def normalize_drop_params(value: object) -> bool | None:
return lax_bool(value)
def drops_unsupported_params(drop_params: object) -> bool:
return litellm.drop_params is True or normalize_drop_params(drop_params) is True

16
litellm/litellm_core_utils/pattern_match_deployments.py
View file

@ -5,6 +5,7 @@ Class to handle llm wildcard routing and regex pattern matching
import copy
import itertools
import re
from collections.abc import Mapping
from re import Match
from typing import Final
@ -22,6 +23,16 @@ class _RoutedDeployment(BaseModel):
_ROUTED_DEPLOYMENTS: Final = TypeAdapter(tuple[_RoutedDeployment, ...])
_PATTERN_WIDE_CACHE_FLAGS: Final = frozenset(("supports_prompt_cache_breakpoint", "supports_prompt_caching"))
_AZURE_AI_PATTERN_CACHE_FLAGS: Final = frozenset(("supports_prompt_caching",))
def _cache_flags_left_to_the_catalog(litellm_params: Mapping[str, object]) -> frozenset[str]:
model: Final = litellm_params.get("model")
provider: Final = litellm_params.get("custom_llm_provider") or inferred_provider(
model if isinstance(model, str) else None
)
return _AZURE_AI_PATTERN_CACHE_FLAGS if provider == "azure_ai" else _PATTERN_WIDE_CACHE_FLAGS
class PatternUtils:
@ -133,6 +144,11 @@ class PatternMatchRouter:
matched_pattern=matched_pattern,
litellm_deployment_litellm_model=deployment["litellm_params"]["model"],
)
if "*" in deployment["litellm_params"]["model"] and isinstance(new_deployment.get("model_info"), dict):
left_to_the_catalog = _cache_flags_left_to_the_catalog(deployment["litellm_params"])
new_deployment["model_info"] = {
key: value for key, value in new_deployment["model_info"].items() if key not in left_to_the_catalog
}
new_deployments.append(new_deployment)
return new_deployments

52
litellm/litellm_core_utils/prompt_templates/factory.py
View file

@ -4254,6 +4254,7 @@ def _convert_to_bedrock_tool_call_invoke(
tool_calls: list,
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> list[BedrockContentBlock]:
"""
OpenAI tool invokes:
@ -4338,6 +4339,7 @@ def _convert_to_bedrock_tool_call_invoke(
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None:
_parts_list.append(_cache_point_block)
@ -4358,6 +4360,7 @@ def _convert_to_bedrock_tool_call_invoke(
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if cache_point_block is not None:
_parts_list.append(cache_point_block)
@ -5029,6 +5032,7 @@ class BedrockConverseMessagesProcessor:
deployment_model: str | None = None,
base_model: str | None = None,
drop_params: bool = False,
model_info: Mapping[str, object] | None = None,
) -> list[BedrockMessageBlock]:
from litellm.llms.bedrock.common_utils import bedrock_converse_accepts_redacted_thinking
@ -5091,6 +5095,7 @@ class BedrockConverseMessagesProcessor:
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None:
_parts.append(_cache_point_block)
@ -5098,7 +5103,11 @@ class BedrockConverseMessagesProcessor:
elif message_content and isinstance(message_content, str):
_part = BedrockContentBlock(text=message_content)
_cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block(
message_block, block_type="content_block", model=model, deployment_model=deployment_model
message_block,
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
user_content.append(_part)
if _cache_point_block is not None:
@ -5149,6 +5158,7 @@ class BedrockConverseMessagesProcessor:
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if cache_point_block is not None:
tool_content.append(cache_point_block)
@ -5214,6 +5224,7 @@ class BedrockConverseMessagesProcessor:
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None and not (
drop_params and _as_thinking_block(element) is not None
@ -5231,6 +5242,7 @@ class BedrockConverseMessagesProcessor:
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None:
assistant_content.append(_cache_point_block)
@ -5239,7 +5251,7 @@ class BedrockConverseMessagesProcessor:
if _tool_calls:
assistant_content.extend(
_convert_to_bedrock_tool_call_invoke(
_tool_calls, model=model, deployment_model=deployment_model
_tool_calls, model=model, deployment_model=deployment_model, model_info=model_info
)
)
@ -5434,6 +5446,7 @@ def _bedrock_converse_messages_pt(
deployment_model: str | None = None,
base_model: str | None = None,
drop_params: bool = False,
model_info: Mapping[str, object] | None = None,
) -> list[BedrockMessageBlock]:
"""
Converts given messages from OpenAI format to Bedrock format
@ -5502,6 +5515,7 @@ def _bedrock_converse_messages_pt(
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None:
_parts.append(_cache_point_block)
@ -5509,7 +5523,11 @@ def _bedrock_converse_messages_pt(
elif message_content and isinstance(message_content, str):
_part = BedrockContentBlock(text=message_content)
_cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block(
message_block, block_type="content_block", model=model, deployment_model=deployment_model
message_block,
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
user_content.append(_part)
if _cache_point_block is not None:
@ -5562,6 +5580,7 @@ def _bedrock_converse_messages_pt(
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if cache_point_block is not None:
tool_content.append(cache_point_block)
@ -5628,6 +5647,7 @@ def _bedrock_converse_messages_pt(
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None and not (
drop_params and _as_thinking_block(element) is not None
@ -5640,14 +5660,20 @@ def _bedrock_converse_messages_pt(
assistant_content.append(BedrockContentBlock(text=_assistant_content))
# Add cache point block for assistant string content
_cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block(
assistant_message_block, block_type="content_block", model=model, deployment_model=deployment_model
assistant_message_block,
block_type="content_block",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if _cache_point_block is not None:
assistant_content.append(_cache_point_block)
_tool_calls = assistant_message_block.get("tool_calls", [])
if _tool_calls:
assistant_content.extend(
_convert_to_bedrock_tool_call_invoke(_tool_calls, model=model, deployment_model=deployment_model)
_convert_to_bedrock_tool_call_invoke(
_tool_calls, model=model, deployment_model=deployment_model, model_info=model_info
)
)
msg_i += 1
@ -5703,7 +5729,10 @@ def make_valid_bedrock_tool_name(input_tool_name: str) -> str:
def add_cache_point_tool_block(
tool: dict, model: str | None = None, deployment_model: str | None = None
tool: dict,
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> BedrockToolBlock | None:
from litellm.llms.bedrock.common_utils import (
bedrock_cache_point_ttl_supported,
@ -5711,7 +5740,7 @@ def add_cache_point_tool_block(
)
cache_control: Final = tool.get("cache_control", None)
if cache_control is not None and bedrock_model_accepts_cache_points(model, deployment_model):
if cache_control is not None and bedrock_model_accepts_cache_points(model, deployment_model, model_info):
cache_point: Final = cache_control.get("type", "ephemeral")
if cache_point == "ephemeral":
cache_point_block: Final[CachePointBlock] = {"type": "default"}
@ -5748,7 +5777,10 @@ def _is_bedrock_tool_block(tool: dict) -> bool:
def _bedrock_tools_pt(
tools: list, model: str | None = None, deployment_model: str | None = None
tools: list,
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> list[BedrockToolBlock]:
"""
OpenAI tools looks like:
@ -5869,7 +5901,9 @@ def _bedrock_tools_pt(
tool_block_list.append(tool_block)
## ADD CACHE POINT TOOL BLOCK ##
cache_point_tool_block = add_cache_point_tool_block(tool, model=model, deployment_model=deployment_model)
cache_point_tool_block = add_cache_point_tool_block(
tool, model=model, deployment_model=deployment_model, model_info=model_info
)
if cache_point_tool_block is not None:
tool_block_list.append(cache_point_tool_block)

26
litellm/llms/anthropic/common_utils.py
View file

@ -28,7 +28,7 @@ from litellm.constants import (
DEFAULT_REASONING_EFFORT_XHIGH_THINKING_BUDGET,
)
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.core_helpers import as_str_mapping, lax_bool
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_file_ids_from_messages,
is_encrypted_reasoning_block,
@ -307,9 +307,10 @@ def count_tokens_feature_betas(
)
def supports_anthropic_cache_control(model: str, custom_llm_provider: str | None) -> bool:
def supports_anthropic_cache_control(
model: str, custom_llm_provider: str | None, model_info: Mapping[str, object] | None = None
) -> bool:
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.utils import supports_prompt_caching
try:
provider: Final = custom_llm_provider if custom_llm_provider is not None else get_llm_provider(model=model)[1]
@ -318,10 +319,27 @@ def supports_anthropic_cache_control(model: str, custom_llm_provider: str | None
return (
provider in ("anthropic", "bedrock", "vertex_ai", "azure_ai")
and "claude" in model.lower()
and supports_prompt_caching(model=model, custom_llm_provider=provider)
and _deployment_supports_prompt_caching(model, provider, model_info)
)
def _deployment_supports_prompt_caching(model: str, provider: str, model_info: Mapping[str, object] | None) -> bool:
from litellm.utils import catalog_with_runtime_registrations, get_model_info, supports_prompt_caching
declared: Final = None if model_info is None else lax_bool(model_info.get("supports_prompt_caching"))
if declared is not None:
return declared
live: Final = supports_prompt_caching(model=model, custom_llm_provider=provider)
if model_info is None:
return live
try:
key: Final = get_model_info(model=model, custom_llm_provider=provider).get("key")
except Exception: # noqa: BLE001 # an unmapped model keeps the live answer
return live
catalog_entry: Final = catalog_with_runtime_registrations().get(key)
return live if catalog_entry is None else catalog_entry.get("supports_prompt_caching") is True
def _tool_config_point_control(points: object) -> Mapping[object, object] | None:
point_fields: Final = tuple(validated_object_mapping(point) for point in validated_object_list(points) or ())
tool_config_point: Final = next(

7
litellm/llms/anthropic/messages/responses_adapters/handler.py
View file

@ -13,6 +13,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
routed_model_info,
supports_openai_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.asyncify import (
@ -187,6 +188,7 @@ def _reads_prompt_cache_breakpoint(model: str, forwarded_kwargs: Mapping[str, ob
provider if isinstance(provider, str) else None,
forwarded_kwargs.get("api_base") or forwarded_kwargs.get("base_url"),
forwarded_kwargs.get("prompt_cache_options"),
routed_model_info(forwarded_kwargs.get("model_info")),
)
@ -194,10 +196,11 @@ def _model_takes_prompt_cache_breakpoint(model: str, forwarded_kwargs: Mapping[s
provider: Final = forwarded_kwargs.get("custom_llm_provider")
provider_name: Final = provider if isinstance(provider, str) else None
base_model: Final = forwarded_kwargs.get("base_model")
return supports_openai_prompt_cache_breakpoint(model, provider_name) or (
model_info: Final = routed_model_info(forwarded_kwargs.get("model_info"))
return supports_openai_prompt_cache_breakpoint(model, provider_name, model_info) or (
isinstance(base_model, str)
and bool(base_model)
and supports_openai_prompt_cache_breakpoint(base_model, provider_name)
and supports_openai_prompt_cache_breakpoint(base_model, provider_name, model_info)
)

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

@ -10,7 +10,11 @@ from pydantic import TypeAdapter, ValidationError
import litellm
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
declared_prompt_cache_breakpoint,
routed_model_info,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import (
content_parts,
convert_content_list_to_str,
@ -125,18 +129,14 @@ def _with_breakpoints_for_foundry(message: AllMessageValues, drop_params: bool)
def _deployment_reads_breakpoints(model: str, litellm_params: Mapping[str, object]) -> bool:
model_info: Final = routed_model_info(litellm_params.get("model_info"))
declared: Final = declared_prompt_cache_breakpoint(model_info)
if declared is not None:
return declared
base_model: Final = litellm_params.get("base_model")
named_models: Final = (base_model, model) if isinstance(base_model, str) and base_model else (model,)
model_info: Final = litellm_params.get("model_info")
if not isinstance(model_info, Mapping):
return any(
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(name, "azure_ai") for name in named_models
)
deployment_flag: Final = _FIELDS.validate_python(model_info).get("supports_prompt_cache_breakpoint")
if isinstance(deployment_flag, bool):
return deployment_flag
return any(
AnthropicCacheControlHook.catalog_targets_openai_prompt_cache_breakpoint(name, "azure_ai")
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(name, "azure_ai", model_info=model_info)
for name in named_models
)

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

@ -1420,6 +1420,7 @@ class AmazonConverseConfig(BaseConfig):
block_type: Literal["system"],
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> SystemContentBlock | None:
pass
@ -1433,6 +1434,7 @@ class AmazonConverseConfig(BaseConfig):
block_type: Literal["content_block"],
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> ContentBlock | None:
pass
@ -1445,9 +1447,10 @@ class AmazonConverseConfig(BaseConfig):
block_type: Literal["system", "content_block"],
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> SystemContentBlock | ContentBlock | None:
cache_control: Final = message_block.get("cache_control", None)
if cache_control is None or not bedrock_model_accepts_cache_points(model, deployment_model):
if cache_control is None or not bedrock_model_accepts_cache_points(model, deployment_model, model_info):
return None
cache_point: Final = self._build_cache_point_block(cache_control, model, deployment_model)
@ -1593,7 +1596,11 @@ class AmazonConverseConfig(BaseConfig):
return ChatCompletionTextObject(type="text", text=text, cache_control=cache_control)
def _transform_system_message(
self, messages: list[AllMessageValues], model: str | None = None, deployment_model: str | None = None
self,
messages: list[AllMessageValues],
model: str | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> tuple[list[AllMessageValues], list[SystemContentBlock]]:
leading_count: Final = next(
(i for i, m in enumerate(messages) if not is_system_message(m)),
@ -1609,7 +1616,7 @@ class AmazonConverseConfig(BaseConfig):
if isinstance(content, str) and content:
system_content_blocks.append(SystemContentBlock(text=content))
cache_block = self.get_cache_point_block(
message, block_type="system", model=model, deployment_model=deployment_model
message, block_type="system", model=model, deployment_model=deployment_model, model_info=model_info
)
if cache_block:
system_content_blocks.append(cache_block)
@ -1618,7 +1625,11 @@ class AmazonConverseConfig(BaseConfig):
if m.get("type") == "text" and m.get("text"):
system_content_blocks.append(SystemContentBlock(text=m["text"]))
cache_block = self.get_cache_point_block(
m, block_type="system", model=model, deployment_model=deployment_model
m,
block_type="system",
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
if cache_block:
system_content_blocks.append(cache_block)
@ -1816,6 +1827,7 @@ class AmazonConverseConfig(BaseConfig):
additional_request_params: dict,
litellm_params: Mapping[str, object] | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> tuple[list[ToolBlock], list]:
"""Process tools and collect anthropic_beta values."""
bedrock_tools: list[ToolBlock] = []
@ -1877,6 +1889,7 @@ class AmazonConverseConfig(BaseConfig):
_tools_the_model_accepts(regular_tools, model, litellm_params),
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
# Add computer use tools and anthropic_beta if needed (only when computer use tools are present)
@ -1946,6 +1959,7 @@ class AmazonConverseConfig(BaseConfig):
_tools_the_model_accepts(_TOOLS_AS_SENT.validate_python(filtered_tools), model, litellm_params),
model=model,
deployment_model=deployment_model,
model_info=model_info,
)
# Append pre-formatted tools (systemTool etc.) after transformation
@ -2021,6 +2035,7 @@ class AmazonConverseConfig(BaseConfig):
drop_params: bool = False,
litellm_params: Mapping[str, object] | None = None,
deployment_model: str | None = None,
model_info: Mapping[str, object] | None = None,
) -> CommonRequestObject:
# Drop thinking param if thinking is enabled but thinking_blocks are missing
# This prevents the error: "Expected thinking or redacted_thinking, but found tool_use"
@ -2059,7 +2074,7 @@ class AmazonConverseConfig(BaseConfig):
# Process tools and collect beta values
bedrock_tools, anthropic_beta_list = self._process_tools_and_beta(
original_tools, model, headers, additional_request_params, litellm_params, deployment_model
original_tools, model, headers, additional_request_params, litellm_params, deployment_model, model_info
)
requested_thinking: Final = additional_request_params.get("thinking") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # _prepare_request_params hands back a bare dict
body_betas: Final = additional_request_params.get("anthropic_beta") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # same bare dict
@ -2077,7 +2092,7 @@ class AmazonConverseConfig(BaseConfig):
if (
cache_injection_points
and len(bedrock_tools) > 0
and bedrock_model_accepts_cache_points(model, deployment_model)
and bedrock_model_accepts_cache_points(model, deployment_model, model_info)
):
for point in cache_injection_points:
if point.get("location") == "tool_config":
@ -2163,8 +2178,11 @@ class AmazonConverseConfig(BaseConfig):
litellm_params: dict[str, object],
headers: dict | None = None,
) -> RequestObject:
from litellm.integrations.anthropic_cache_control_hook import routed_model_info
base_model: Final = litellm_params.get("base_model")
capability_model: Final = bedrock_capability_model(model, base_model)
model_info: Final = routed_model_info(litellm_params.get("model_info"))
messages, system_content_blocks = self._transform_system_message(
list(
self._claude_messages_within_cache_control_cap(
@ -2173,6 +2191,7 @@ class AmazonConverseConfig(BaseConfig):
),
model=capability_model,
deployment_model=model,
model_info=model_info,
)
# Convert last user message to guarded_text if guardrailConfig is present
@ -2184,6 +2203,7 @@ class AmazonConverseConfig(BaseConfig):
_data: Final[CommonRequestObject] = self._transform_request_helper(
model=capability_model,
deployment_model=model,
model_info=model_info,
system_content_blocks=system_content_blocks,
optional_params=optional_params,
messages=messages,
@ -2198,6 +2218,7 @@ class AmazonConverseConfig(BaseConfig):
messages=messages,
model=capability_model,
deployment_model=model,
model_info=model_info,
llm_provider="bedrock_converse",
user_continue_message=litellm_params.pop("user_continue_message", None),
base_model=self._deployment_base_model(_DEPLOYMENT_PARAMS.validate_python(litellm_params)),
@ -2250,8 +2271,11 @@ class AmazonConverseConfig(BaseConfig):
litellm_params: dict[str, object],
headers: dict | None = None,
) -> RequestObject:
from litellm.integrations.anthropic_cache_control_hook import routed_model_info
base_model: Final = litellm_params.get("base_model")
capability_model: Final = bedrock_capability_model(model, base_model)
model_info: Final = routed_model_info(litellm_params.get("model_info"))
messages, system_content_blocks = self._transform_system_message(
list(
self._claude_messages_within_cache_control_cap(
@ -2260,6 +2284,7 @@ class AmazonConverseConfig(BaseConfig):
),
model=capability_model,
deployment_model=model,
model_info=model_info,
)
# Convert last user message to guarded_text if guardrailConfig is present
@ -2270,6 +2295,7 @@ class AmazonConverseConfig(BaseConfig):
_data: Final[CommonRequestObject] = self._transform_request_helper(
model=capability_model,
deployment_model=model,
model_info=model_info,
system_content_blocks=system_content_blocks,
optional_params=optional_params,
messages=messages,
@ -2285,6 +2311,7 @@ class AmazonConverseConfig(BaseConfig):
messages=messages,
model=capability_model,
deployment_model=model,
model_info=model_info,
llm_provider="bedrock_converse",
user_continue_message=litellm_params.pop("user_continue_message", None),
base_model=self._deployment_base_model(_DEPLOYMENT_PARAMS.validate_python(litellm_params)),

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

@ -33,6 +33,7 @@ from pydantic import ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm import verbose_logger
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.core_helpers import lax_bool
from litellm.litellm_core_utils.optional_imports import ensure_optional_import
from litellm.litellm_core_utils.prompt_templates.common_utils import schema_property_count
from litellm.litellm_core_utils.url_utils import ensure_api_route
@ -1515,7 +1516,9 @@ def bedrock_converse_supports_parallel_tool_use_config(model: str) -> bool:
)
def bedrock_model_accepts_cache_points(model: str | None, deployment_model: str | None = None) -> bool:
def bedrock_model_accepts_cache_points(
model: str | None, deployment_model: str | None = None, model_info: Mapping[str, object] | None = None
) -> bool:
"""
Whether Converse ``cachePoint`` blocks may be sent to this model.
@ -1535,43 +1538,61 @@ def bedrock_model_accepts_cache_points(model: str | None, deployment_model: str
never silently degrade. ``litellm.utils.supports_prompt_caching`` is not reusable here:
it returns False for unmapped models, the opposite polarity. ``deployment_model`` is the
model as configured when ``model`` is its ``base_model``: its own declaration wins.
A routed call passes its deployment's ``model_info``, whose breakpoint flag wins outright and
whose ``supports_prompt_caching`` wins over the catalog's for a known model. Other flags come
from the catalog as loaded, since the registered keys are shared by every deployment of the
same model string and carry whichever sibling registered last.
"""
if model is None:
return True
deployment_declared: Final = (
None if deployment_model is None else bedrock_declared_cache_point_support(deployment_model)
None if deployment_model is None else bedrock_declared_cache_point_support(deployment_model, model_info)
)
declared: Final = (
deployment_declared if deployment_declared is not None else bedrock_declared_cache_point_support(model)
deployment_declared
if deployment_declared is not None
else bedrock_declared_cache_point_support(model, model_info)
)
return declared is not False
def bedrock_declared_cache_point_support(model: str) -> bool | None:
def bedrock_declared_cache_point_support(model: str, model_info: Mapping[str, object] | None = None) -> bool | None:
if _OPENAI_FAMILY_MODEL_RE.search(model):
return False
deployment_flag: Final = None if model_info is None else model_info.get("supports_prompt_cache_breakpoint")
if deployment_flag is not None:
return deployment_flag is True
catalog_entry: Final = _catalog_entry_reader(model_info)
map_keys: Final = (model, get_bedrock_base_model(model))
registered_keys: Final = tuple(f"bedrock/{route}{model}" for route in ("", "converse/", "converse_like/"))
explicit_marker_support: Final = next(
(
entry.get("supports_prompt_cache_breakpoint") is True
for key in (*registered_keys, *map_keys)
if (entry := litellm.model_cost.get(key)) is not None
and entry.get("supports_prompt_cache_breakpoint") is not None
if (entry := catalog_entry(key)) is not None and entry.get("supports_prompt_cache_breakpoint") is not None
),
None,
)
if explicit_marker_support is not None:
return explicit_marker_support
if not any(key in litellm.model_cost for key in map_keys):
if not any(catalog_entry(key) is not None for key in map_keys):
return None
deployment_caching: Final = None if model_info is None else lax_bool(model_info.get("supports_prompt_caching"))
if deployment_caching is not None:
return deployment_caching
return any(
entry.get("supports_prompt_caching") is True
for key in map_keys
if (entry := litellm.model_cost.get(key)) is not None
entry.get("supports_prompt_caching") is True for key in map_keys if (entry := catalog_entry(key)) is not None
)
def _catalog_entry_reader(model_info: Mapping[str, object] | None) -> Callable[[str], Mapping[str, object] | None]:
from litellm.utils import catalog_with_runtime_registrations
if model_info is not None:
return catalog_with_runtime_registrations().get
return _model_cost_entry
def bedrock_supports_tool_search(model: str) -> bool:
"""
Whether Bedrock InvokeModel admits the ``tool_search_tool_*`` tool types on ``model``.

19
litellm/llms/openai/chat/gpt_transformation.py
View file

@ -556,18 +556,23 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
) -> tuple[Sequence[AllMessageValues], Sequence["ChatCompletionToolParam"] | None]:
"""
A deployment that reads OpenAI's ``prompt_cache_breakpoint`` (gpt-5.6+ on
api.openai.com, or one opted in through ``prompt_cache_options`` or the cost
map) gets each Anthropic ``cache_control`` mark rewritten in place, and under
drop_params each content part's ``prompt_cache_breakpoint`` cut down to what chat
completions accepts. Any other endpoint keeps or strips the marks as before, and
under drop_params also loses every ``prompt_cache_breakpoint`` it cannot read.
api.openai.com, or one opted in through ``prompt_cache_options``, its own
``model_info``, or the cost map) gets each
Anthropic ``cache_control`` mark rewritten in place, and under drop_params each
content part's ``prompt_cache_breakpoint`` cut down to what chat completions
accepts. Any other endpoint keeps or strips the marks as before, and under
drop_params also loses every ``prompt_cache_breakpoint`` it cannot read.
"""
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook, routed_model_info
custom_llm_provider: Final = _litellm_params_str(litellm_params, "custom_llm_provider")
api_base: Final = _litellm_params_str(litellm_params, "api_base")
reads_breakpoints: Final = AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, custom_llm_provider, api_base, prompt_cache_options
model,
custom_llm_provider,
api_base,
prompt_cache_options,
routed_model_info(litellm_params.get("model_info")),
)
if not reads_breakpoints and self._should_preserve_cache_control_for_endpoint(custom_llm_provider, api_base):
return (

2
litellm/llms/openai/responses/transformation.py
View file

@ -12,6 +12,7 @@ import litellm
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
deployment_model_info,
deployment_prompt_cache_breakpoint_support,
targets_openai_api,
)
@ -767,6 +768,7 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
custom_llm_provider=_provider_name(self.custom_llm_provider),
api_base=host.api_base or host.base_url,
prompt_cache_options=prompt_cache_options,
model_info=deployment_model_info(litellm_params),
)
@staticmethod

4
litellm/main.py
View file

@ -755,6 +755,7 @@ async def acompletion(
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
routed_model_info,
)
from litellm.types.llms.openai import AllMessageValues
@ -766,6 +767,7 @@ async def acompletion(
tools=tools,
enable_prompt_caching=cast(bool | None, kwargs.get("enable_prompt_caching")), # cast-ok: untyped kwargs
api_base=base_url or kwargs.get("api_base"),
model_info=routed_model_info(caller_kwargs.get("model_info")),
)
prompt_id: Final = cast(str | None, kwargs.get("prompt_id", None)) # cast-ok: read from untyped kwargs
@ -5827,6 +5829,7 @@ def completion(
from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
routed_model_info,
)
from litellm.types.llms.openai import AllMessageValues
@ -5838,6 +5841,7 @@ def completion(
tools=tools,
enable_prompt_caching=cast(bool | None, kwargs.get("enable_prompt_caching")), # cast-ok: untyped kwargs
api_base=base_url or kwargs.get("api_base"),
model_info=routed_model_info(cast(object, model_info)), # cast-ok: untyped kwargs
)
if isinstance(litellm_logging_obj, LiteLLMLoggingObj) and (

23
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -52,6 +52,7 @@ from litellm.integrations.anthropic_cache_control_hook import (
cache_control_count,
dropped_cache_control,
newest_marks_within_cache_control_limit,
routed_model_info,
whole_messages_within_cache_control_limit,
with_cache_control,
)
@ -782,13 +783,15 @@ def _replays_compaction_block(input: str | ResponseInputParam, context_managemen
_PROVIDERS_KEEPING_MESSAGE_CACHE_CONTROL: Final = frozenset(("anthropic", "dashscope", "minimax", "sap", "zai"))
def _bedrock_target_keeps_item_marks(model: str, drop_params: bool) -> bool:
declared: Final = bedrock_declared_cache_point_support(model)
def _bedrock_target_keeps_item_marks(model: str, drop_params: bool, model_info: Mapping[str, object] | None) -> bool:
declared: Final = bedrock_declared_cache_point_support(model, model_info)
return declared is True or (declared is None and not drop_params)
def _target_keeps_item_marks(model: str, custom_llm_provider: str | None, drop_params: bool) -> bool:
if supports_anthropic_cache_control(model, custom_llm_provider):
def _target_keeps_item_marks(
model: str, custom_llm_provider: str | None, drop_params: bool, model_info: Mapping[str, object] | None
) -> bool:
if supports_anthropic_cache_control(model, custom_llm_provider, model_info):
return True
if custom_llm_provider in _PROVIDERS_KEEPING_MESSAGE_CACHE_CONTROL:
return True
@ -798,7 +801,7 @@ def _target_keeps_item_marks(model: str, custom_llm_provider: str | None, drop_p
return is_snowflake_claude_model(model)
if custom_llm_provider == "databricks":
return "claude" in model.lower()
return custom_llm_provider == "bedrock" and _bedrock_target_keeps_item_marks(model, drop_params)
return custom_llm_provider == "bedrock" and _bedrock_target_keeps_item_marks(model, drop_params, model_info)
def _refuses_marks_on_tool_blocks(model: str, custom_llm_provider: str | None) -> bool:
@ -1234,7 +1237,10 @@ class LiteLLMCompletionResponsesConfig:
request_kwargs: Final = _STR_KEY_DICT_ADAPTER.validate_python({**kwargs, "drop_params": drop_params})
context_management: Final = responses_api_request.get("context_management")
keeps_item_marks: Final = _target_keeps_item_marks(
model, custom_llm_provider, drops_unsupported_params(request_kwargs.get("drop_params"))
model,
custom_llm_provider,
drops_unsupported_params(request_kwargs.get("drop_params")),
routed_model_info(request_kwargs.get("model_info")),
)
request_input: Final = with_compaction_items_as_params(input)
native_compaction: Final = _replays_compaction_block(
@ -1456,7 +1462,10 @@ class LiteLLMCompletionResponsesConfig:
model: Final = str(request_fields.get("model") or "")
provider: Final = requested_provider if isinstance(requested_provider, str) else None
keeps_item_marks: Final = _target_keeps_item_marks(
model, provider, drops_unsupported_params(request_kwargs.get("drop_params"))
model,
provider,
drops_unsupported_params(request_kwargs.get("drop_params")),
routed_model_info(request_kwargs.get("model_info")),
)
lookup: Final = session_lookup or partial(
ResponsesSessionHandler.get_chat_completion_message_history_for_previous_response_id,

2
litellm/responses/main.py
View file

@ -27,6 +27,7 @@ from litellm.integrations.anthropic_cache_control_hook import (
CARRY_UNMATCHED_MESSAGE_POINTS,
AnthropicCacheControlHook,
configured_injection_points,
routed_model_info,
)
from litellm.litellm_core_utils.asyncify import run_async_function
from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind
@ -882,6 +883,7 @@ def _stamp_injection_points_for_the_hook(
custom_llm_provider,
kwargs.get("api_base") or kwargs.get("base_url"),
kwargs.get("prompt_cache_options"),
routed_model_info(kwargs.get("model_info")),
)

60
litellm/utils.py
View file

@ -4226,6 +4226,54 @@ def _register_on_current_map(
_runtime_registered_model_cost: Final[dict[str, dict[str, object]]] = {} # mutable-ok: replayed on reload
_runtime_registration_order: Final[dict[str, int]] = {} # mutable-ok: tells which of a row's registered names came last
_runtime_registered_names: Final[dict[str, tuple[str, ...]]] = {} # mutable-ok: names registered onto each live row
_REGISTRATION_TICKS: Final = itertools.count()
@dataclass(frozen=True, slots=True)
class _RegistrationsOverCatalog(Mapping[str, Mapping[str, object]]):
catalog: Mapping[str, Mapping[str, object]]
registrations: Mapping[str, Mapping[str, object]]
registration_order: Mapping[str, int]
registered_names: Mapping[str, tuple[str, ...]]
def __getitem__(self, key: str) -> Mapping[str, object]:
catalog_entry: Final = self.catalog.get(key)
names: Final = dict.fromkeys((*self.registered_names.get(key, ()), key))
present: Final = tuple(name for name in names if name in self.registrations)
if not present:
return self.catalog[key]
oldest_first: Final = sorted(present, key=lambda name: self.registration_order.get(name, -1))
registered: Final = (self.registrations[name] for name in oldest_first)
return MappingProxyType(
dict(itertools.chain.from_iterable(entry.items() for entry in (catalog_entry or {}, *registered)))
)
def __iter__(self) -> Iterator[str]:
return iter({**dict.fromkeys(self.catalog), **dict.fromkeys(self.registrations)})
def __len__(self) -> int:
return len(self.catalog.keys() | self.registrations.keys())
def _record_registered_name(written_key: str, registered_key: str) -> None:
names: Final = _runtime_registered_names.get(written_key, ())
if registered_key not in names:
_runtime_registered_names[written_key] = (*names, registered_key)
def catalog_with_runtime_registrations() -> Mapping[str, Mapping[str, object]]:
"""The loaded catalog plus ``register_model`` calls, without what routers register for their deployments."""
catalog: Final = GetModelCostMap.loaded_model_cost_map()
if not _runtime_registered_model_cost:
return catalog
return _RegistrationsOverCatalog(
catalog,
MappingProxyType(_runtime_registered_model_cost),
MappingProxyType(_runtime_registration_order),
MappingProxyType(_runtime_registered_names),
)
def runtime_model_cost_registrations(
@ -4291,7 +4339,14 @@ def reapply_runtime_model_cost_registrations() -> None:
if _LiveDeploymentReplay.callback is not None:
_LiveDeploymentReplay.callback()
if _runtime_registered_model_cost:
register_model(model_cost=dict(_runtime_registered_model_cost))
register_model(
model_cost=dict(
sorted(
_runtime_registered_model_cost.items(),
key=lambda registration: _runtime_registration_order.get(registration[0], -1),
)
)
)
def cost_map_omits_token_price(*keys: object) -> bool:
@ -4365,9 +4420,12 @@ def register_model(
dict(_runtime_registered_model_cost.get(_registered_key, {})),
_RATE_TABLE.validate_python(_registered_value),
)
_runtime_registration_order[_registered_key] = next(_REGISTRATION_TICKS)
for key, value in loaded_model_cost.items():
write = _register_on_current_map(str(key), value, custom_llm_provider, warning_display_name)
if persist_across_reloads:
_record_registered_name(write.key, str(key))
# Invalidate case-insensitive lookup map since model_cost was modified
_invalidate_model_cost_lowercase_map()

1
tests/code_coverage_tests/router_code_coverage.py
View file

@ -138,6 +138,7 @@ ignored_function_names = [
"_resolve_pre_call_check_input_tokens", # Tested through the token-count failure pre-call check tests in test_router.py
"_params_neither_dropped_nor_allowed", # Tested through the allowed-or-dropped param pre-call check tests in test_router.py
"_deployment_rejects_params", # Tested through the unsupported-param pre-call check tests in test_router.py
"_cache_flags_left_to_the_catalog", # Tested through the wildcard prompt cache flag tests in test_anthropic_cache_control_hook.py
"_deployment_lacks_params", # Tested through the unsupported-param pre-call check tests in test_router.py
"_pre_call_check_requested_chat_params", # Tested through the unsupported-param pre-call check tests in test_router.py
"_deployment_supports_param", # Tested through the deployment-capability pre-call check tests in test_router.py

10
tests/integration/providers/test_bedrock_kimi_k3_cache_point_wire.py
View file

@ -1251,17 +1251,19 @@ def test_s01_to_s06_an_odd_typed_flag_is_read_as_false_never_as_true(
@pytest.mark.timeout(600)
def test_s10_s11_two_deployments_of_one_arn_share_the_flag_and_a_delete_leaves_it_in_place(gateway: Gateway) -> None:
def test_s10_s11_two_deployments_of_one_arn_keep_their_own_flags_and_a_delete_changes_neither(
gateway: Gateway,
) -> None:
arn: Final = _arn("shared")
with wire_server(_peer()) as wire, gateway.scenario() as scenario:
flagged: Final = _deployment(
gateway, scenario, wire, f"bedrock/{arn}", {"supports_prompt_cache_breakpoint": False}, cleanup=False
)
plain: Final = _deployment(gateway, scenario, wire, f"bedrock/{arn}")
for name in (flagged, plain):
for name, cache_points in ((flagged, 0), (plain, 2)):
outcome, received = _observe_at(gateway, wire, "chat", _chat_body(name, _prompt(), _SYSTEM_AND_USER))
_assert_answered(outcome)
assert received.cache_points == 0, (name, received.body)
assert received.cache_points == cache_points, (name, received.body)
_success_row(outcome.response_id)
flagged_identity: Final = _model_id(flagged)
gateway.post("/model/delete", {"id": flagged_identity})
@ -1276,7 +1278,7 @@ def test_s10_s11_two_deployments_of_one_arn_share_the_flag_and_a_delete_leaves_i
wire.drain()
after, after_received = _observe_at(gateway, wire, "chat", _chat_body(plain, _prompt(), _SYSTEM_AND_USER))
_assert_answered(after)
assert after_received.cache_points == 0, after_received.body
assert after_received.cache_points == 2, after_received.body
_success_row(after.response_id)

32
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
View file

@ -5482,6 +5482,38 @@ def test_transform_request_strips_the_breakpoint_when_the_served_providers_row_o
]
@pytest.mark.parametrize(
("sibling_flag", "model_info", "keeps_the_breakpoint"),
[
(True, {"id": "flagless"}, False),
(True, {"id": "opted-out", "supports_prompt_cache_breakpoint": False}, False),
(False, {"id": "opted-in", "supports_prompt_cache_breakpoint": True}, True),
],
ids=["flagless", "opted-out", "opted-in"],
)
def test_transform_request_reads_the_routed_deployments_own_flag_not_a_siblings_on_the_shared_key(
monkeypatch: pytest.MonkeyPatch,
sibling_flag: bool,
model_info: Mapping[str, object],
keeps_the_breakpoint: bool,
) -> None:
monkeypatch.setitem(
litellm.model_cost,
"bedrock_mantle/qa-mantle-shared",
{**_MANTLE_GPT_ROW, "supports_prompt_cache_breakpoint": sibling_flag},
)
system_content: Final = _bridged_input_for_served_provider(
"qa-mantle-shared", "bedrock_mantle", model_info=model_info
)[0]["content"]
assert system_content == [
{**_MARKED_SYSTEM_PART, "type": "input_text"}
if keeps_the_breakpoint
else {"type": "input_text", "text": "Stable prefix"}
]
def test_transform_request_strips_the_breakpoint_for_a_served_provider_without_a_row_of_its_own(
monkeypatch: pytest.MonkeyPatch,
) -> None:

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

@ -8,14 +8,18 @@ import subprocess
import sys
import textwrap
from collections.abc import Mapping
from types import MappingProxyType
from typing import Final, List, Optional, Tuple, cast
from unittest.mock import MagicMock, patch
import pytest
import respx
from httpx import Response
from openai.types.chat import ChatCompletionMessage
from pydantic import BaseModel, ConfigDict
import litellm
from litellm import Router
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import (
CARRY_UNMATCHED_MESSAGE_POINTS,
@ -5055,6 +5059,635 @@ def test_cache_control_hook_injects_next_to_a_malformed_message_instead_of_raisi
]
_SHARED_FOUNDRY_MODEL: Final = "azure_ai/qa-shared-gpt-deployment"
_SIBLING_BREAKPOINT_FLAGS: Final = MappingProxyType(
{
"opted-in": MappingProxyType({"supports_prompt_cache_breakpoint": True}),
"opted-out": MappingProxyType({"supports_prompt_cache_breakpoint": False}),
}
)
_FOUNDRY_CHAT_ROUTE: Final = (
"/models/chat/completions",
MappingProxyType(
{
"id": "chatcmpl-azure-ai",
"object": "chat.completion",
"created": 1700000000,
"model": "qa-shared-gpt-deployment",
"choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 5, "completion_tokens": 1, "total_tokens": 6},
}
),
)
_FOUNDRY_RESPONSES_ROUTE: Final = (
"/openai/v1/responses",
MappingProxyType(
{
"id": "resp_azure_ai",
"object": "response",
"created_at": 1700000000,
"status": "completed",
"model": "qa-shared-gpt-deployment",
"output": [
{
"type": "message",
"id": "msg_azure_ai",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "ok", "annotations": []}],
}
],
}
),
)
def _foundry_siblings_sharing_one_model_string(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
registration_order: tuple[str, ...],
upstream: tuple[str, Mapping[str, object]],
) -> tuple[Router, Mapping[str, respx.Route]]:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
path, reply = upstream
routes: Final = MappingProxyType(
{
host: respx_mock.post(host=f"{host}.services.ai.azure.com", path=path).mock(
return_value=Response(200, json=dict(reply))
)
for host in registration_order
}
)
router: Final = Router(
model_list=[
{
"model_name": host,
"litellm_params": {
"model": _SHARED_FOUNDRY_MODEL,
"api_base": f"https://{host}.services.ai.azure.com/models",
"api_key": "azure-ai-test-key",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": f"{host}-id", **_SIBLING_BREAKPOINT_FLAGS.get(host, {})},
}
for host in registration_order
],
num_retries=0,
)
return router, routes
async def _send_unmarked_system_prompt(router: Router, host: str, entrypoint: str) -> None:
chat_messages: Final = [{"role": "system", "content": "stable"}, {"role": "user", "content": "hi"}]
match entrypoint:
case "completion":
router.completion(model=host, messages=chat_messages)
case "acompletion":
await router.acompletion(model=host, messages=chat_messages)
case "aanthropic_messages":
await router.aanthropic_messages(
model=host, system="stable", messages=[{"role": "user", "content": "hi"}], max_tokens=16
)
case "aresponses":
await router.aresponses(model=host, input=chat_messages)
@pytest.mark.parametrize(
("entrypoint", "upstream"),
[
("completion", _FOUNDRY_CHAT_ROUTE),
("acompletion", _FOUNDRY_CHAT_ROUTE),
("aanthropic_messages", _FOUNDRY_CHAT_ROUTE),
("aresponses", _FOUNDRY_RESPONSES_ROUTE),
],
ids=["completion", "acompletion", "aanthropic_messages", "aresponses"],
)
@pytest.mark.parametrize(
"registration_order", [("opted-in", "opted-out", "unflagged"), ("unflagged", "opted-out", "opted-in")]
)
async def test_foundry_deployment_reads_only_its_own_prompt_cache_flag_not_a_siblings_sharing_its_model_string(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
registration_order: tuple[str, ...],
entrypoint: str,
upstream: tuple[str, Mapping[str, object]],
) -> None:
router, routes = _foundry_siblings_sharing_one_model_string(respx_mock, monkeypatch, registration_order, upstream)
for host in registration_order:
await _send_unmarked_system_prompt(router, host, entrypoint)
def sent_prompt_caching(host: str) -> tuple[object, bool]:
body: Final = routes[host].calls.last.request.content
return json.loads(body).get("prompt_cache_options"), b"prompt_cache_breakpoint" in body
assert {host: sent_prompt_caching(host) for host in registration_order} == {
"opted-in": ({"mode": "implicit"}, True),
"opted-out": (None, False),
"unflagged": (None, False),
}
@pytest.mark.parametrize(
("model", "sibling_flag", "expected"),
[("qa-openai-shared", True, False), ("gpt-99.0-qa-shared", False, True)],
ids=["sibling-opt-in-does-not-turn-it-on", "sibling-opt-out-does-not-turn-it-off"],
)
def test_openai_deployment_reads_the_catalog_not_a_siblings_flag_on_the_shared_key(
monkeypatch: pytest.MonkeyPatch, model: str, sibling_flag: bool, expected: bool
) -> None:
sibling_entry: Final = {
"litellm_provider": "openai",
"mode": "chat",
"supports_prompt_cache_breakpoint": sibling_flag,
}
monkeypatch.setitem(litellm.model_cost, model, sibling_entry)
assert (
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, "openai", "https://api.openai.com/v1", model_info={"id": "flagless-deployment"}
)
is expected
)
@pytest.mark.parametrize(
("model", "declared", "api_base", "expected"),
[
("qa-openai-shared", True, "https://api.openai.com/v1", True),
("qa-openai-shared", True, "https://compatible-host.example.com/v1", False),
("gpt-99.0-qa-shared", False, "https://api.openai.com/v1", False),
],
ids=["opt-in-on-openai", "opt-in-on-a-compatible-host-stays-off", "opt-out-beats-the-version-rule"],
)
def test_openai_deployment_declared_flag_answers_support_while_the_host_still_decides(
model: str, declared: bool, api_base: str, expected: bool
) -> None:
assert (
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, "openai", api_base, model_info={"supports_prompt_cache_breakpoint": declared}
)
is expected
)
@pytest.mark.parametrize(
"registration_order", [("opted-in", "opted-out", "unflagged"), ("unflagged", "opted-out", "opted-in")]
)
async def test_foundry_responses_drop_params_keeps_a_client_breakpoint_by_the_deployments_own_flag(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, registration_order: tuple[str, ...]
) -> None:
router, routes = _foundry_siblings_sharing_one_model_string(
respx_mock, monkeypatch, registration_order, _FOUNDRY_RESPONSES_ROUTE
)
client_marked_system_prompt: Final = {
"role": "system",
"content": [{"type": "input_text", "text": "stable", "prompt_cache_breakpoint": {"mode": "explicit"}}],
}
for host in registration_order:
await router.aresponses(
model=host, input=[client_marked_system_prompt, {"role": "user", "content": "hi"}], drop_params=True
)
assert {
host: b"prompt_cache_breakpoint" in routes[host].calls.last.request.content for host in registration_order
} == {
"opted-in": True,
"opted-out": False,
"unflagged": True,
}
@pytest.mark.parametrize(
("entrypoint", "part_type", "reply"),
[("acompletion", "text", _FOUNDRY_CHAT_ROUTE[1]), ("aresponses", "input_text", _FOUNDRY_RESPONSES_ROUTE[1])],
ids=["chat", "responses"],
)
@pytest.mark.parametrize("registration_order", [("opted-in", "unflagged"), ("unflagged", "opted-in")])
async def test_azure_drop_params_keeps_a_client_breakpoint_by_the_deployments_own_flag_not_a_siblings(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
registration_order: tuple[str, ...],
entrypoint: str,
part_type: str,
reply: Mapping[str, object],
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
routes: Final = MappingProxyType(
{
host: respx_mock.post(host=f"{host}.openai.azure.com").mock(return_value=Response(200, json=dict(reply)))
for host in registration_order
}
)
router: Final = Router(
model_list=[
{
"model_name": host,
"litellm_params": {
"model": "azure/gpt-4.1-qa-shared",
"api_base": f"https://{host}.openai.azure.com",
"api_key": "azure-test-key",
"api_version": "2025-04-01-preview",
},
"model_info": {"id": f"{host}-id", **_SIBLING_BREAKPOINT_FLAGS.get(host, {})},
}
for host in registration_order
],
num_retries=0,
)
marked_system_prompt: Final = {
"role": "system",
"content": [{"type": part_type, "text": "stable", "prompt_cache_breakpoint": {"mode": "explicit"}}],
}
request: Final = [marked_system_prompt, {"role": "user", "content": "hi"}]
for host in registration_order:
match entrypoint:
case "acompletion":
await router.acompletion(model=host, messages=request, drop_params=True)
case "aresponses":
await router.aresponses(model=host, input=request, drop_params=True)
assert {
host: b"prompt_cache_breakpoint" in routes[host].calls.last.request.content for host in registration_order
} == {"opted-in": True, "unflagged": False}
@pytest.mark.parametrize("registration_order", [("opted-in", "unflagged"), ("unflagged", "opted-in")])
async def test_openai_messages_bridge_sends_a_breakpoint_by_the_deployments_own_flag_not_a_siblings(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, registration_order: tuple[str, ...]
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
_, reply = _FOUNDRY_RESPONSES_ROUTE
responses_route: Final = respx_mock.post("https://api.openai.com/v1/responses").mock(
return_value=Response(200, json=dict(reply))
)
router: Final = Router(
model_list=[
{
"model_name": name,
"litellm_params": {"model": "openai/qa-shared-gpt-deployment", "api_key": "openai-test-key"},
"model_info": {"id": f"{name}-id", **_SIBLING_BREAKPOINT_FLAGS.get(name, {})},
}
for name in registration_order
],
num_retries=0,
)
async def sends_a_breakpoint(name: str) -> bool:
await router.aanthropic_messages(
model=name,
system=[{"type": "text", "text": "stable", "cache_control": {"type": "ephemeral"}}],
messages=[{"role": "user", "content": "hi"}],
max_tokens=16,
)
return b"prompt_cache_breakpoint" in responses_route.calls.last.request.content
assert {name: await sends_a_breakpoint(name) for name in registration_order} == {
"opted-in": True,
"unflagged": False,
}
_OPENAI_ROUTES: Final = MappingProxyType(
{
"acompletion": ("https://api.openai.com/v1/chat/completions", _FOUNDRY_CHAT_ROUTE[1]),
"aresponses": ("https://api.openai.com/v1/responses", _FOUNDRY_RESPONSES_ROUTE[1]),
"aanthropic_messages": ("https://api.openai.com/v1/responses", _FOUNDRY_RESPONSES_ROUTE[1]),
}
)
def _openai_route(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, entrypoint: str) -> respx.Route:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
url, reply = _OPENAI_ROUTES[entrypoint]
return respx_mock.post(url).mock(return_value=Response(200, json=dict(reply)))
@pytest.mark.parametrize("declared", [True, False])
async def test_a_wildcard_deployments_prompt_cache_flag_leaves_each_matched_model_to_the_catalog(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, declared: bool
) -> None:
chat: Final = _openai_route(respx_mock, monkeypatch, "acompletion")
router: Final = Router(
model_list=[
{
"model_name": "openai/*",
"litellm_params": {
"model": "openai/*",
"api_key": "openai-test-key",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": "wildcard-id", "supports_prompt_cache_breakpoint": declared},
}
],
num_retries=0,
)
async def sends_prompt_cache_options(model: str) -> bool:
await _send_unmarked_system_prompt(router, model, "acompletion")
return b"prompt_cache_options" in chat.calls.last.request.content
assert {model: await sends_prompt_cache_options(model) for model in ("openai/gpt-4.1-qa", "openai/gpt-99.0-qa")} == {
"openai/gpt-4.1-qa": False,
"openai/gpt-99.0-qa": True,
}
async def test_an_azure_ai_wildcard_deployments_breakpoint_flag_still_speaks_for_each_matched_model(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
path, reply = _FOUNDRY_CHAT_ROUTE
chat: Final = respx_mock.post(host="qa-wildcard.services.ai.azure.com", path=path).mock(
return_value=Response(200, json=dict(reply))
)
router: Final = Router(
model_list=[
{
"model_name": "azure_ai/*",
"litellm_params": {
"model": "azure_ai/*",
"api_base": "https://qa-wildcard.services.ai.azure.com/models",
"api_key": "azure-ai-test-key",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": "wildcard-id", "supports_prompt_cache_breakpoint": True},
}
],
num_retries=0,
)
await _send_unmarked_system_prompt(router, "azure_ai/qa-unlisted-alias", "acompletion")
body: Final = chat.calls.last.request.content
assert (json.loads(body).get("prompt_cache_options"), b"prompt_cache_breakpoint" in body) == (
{"mode": "implicit"},
True,
)
@pytest.mark.parametrize("entrypoint", ["acompletion", "aresponses", "aanthropic_messages"])
@pytest.mark.parametrize("declared_in", ["litellm_params", "model_info"])
async def test_a_base_model_that_takes_breakpoints_leaves_an_openai_deployments_request_as_it_was(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, entrypoint: str, declared_in: str
) -> None:
upstream: Final = _openai_route(respx_mock, monkeypatch, entrypoint)
def deployment(name: str, base_model: Mapping[str, str]) -> dict[str, object]:
return {
"model_name": name,
"litellm_params": {
"model": "openai/gpt-4.1-qa",
"api_key": "openai-test-key",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
**(base_model if declared_in == "litellm_params" else {}),
},
"model_info": {"id": f"{name}-id", **(base_model if declared_in == "model_info" else {})},
}
router: Final = Router(
model_list=[deployment("plain", {}), deployment("priced-as-breakpoint-model", {"base_model": "gpt-99.0-qa"})],
num_retries=0,
)
async def sent_body(name: str) -> object:
await _send_unmarked_system_prompt(router, name, entrypoint)
return json.loads(upstream.calls.last.request.content)
assert await sent_body("priced-as-breakpoint-model") == await sent_body("plain")
def test_a_routed_call_honors_a_register_model_breakpoint_opt_out_made_under_the_provider_prefixed_name() -> None:
from litellm.utils import catalog_with_runtime_registrations
model: Final = next(
key
for key, entry in catalog_with_runtime_registrations().items()
if entry.get("litellm_provider") == "openai" and entry.get("supports_prompt_cache_breakpoint") is True
)
litellm.register_model({f"openai/{model}": {"supports_prompt_cache_breakpoint": False}})
assert supports_openai_prompt_cache_breakpoint(model, "openai", {"id": "plain-id"}) is False
_APPLICATION_INFERENCE_PROFILE: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/qa"
@pytest.mark.parametrize(
("model", "custom_llm_provider", "expected"),
[
("us.anthropic.claude-sonnet-5-5", "bedrock", False),
("converse/us.anthropic.claude-sonnet-5-5", "bedrock", False),
("invoke/us.anthropic.claude-sonnet-5-5", "bedrock", False),
("converse_like/us.anthropic.claude-sonnet-5-5", "bedrock", False),
("openai/gpt-oss-120b", "bedrock", False),
("moonshotai.kimi-k3", "bedrock", False),
(_APPLICATION_INFERENCE_PROFILE, "bedrock", False),
("claude-sonnet-5-5", "anthropic", False),
("claude-sonnet-5-5", "azure_ai", False),
("claude-sonnet-5-5", "vertex_ai", False),
("gemini-3-pro", "gemini", False),
("qa-custom-model", "qa-custom-handler", False),
("qa-shared-gpt-deployment", "azure_ai", True),
("my-custom-deployment", "azure", True),
("qa-mantle-model", "bedrock_mantle", True),
("anthropic/qa-gateway-claude", "openrouter", False),
("qa-downstream-claude", "litellm_proxy", False),
("qa-self-hosted-model", "hosted_vllm", False),
],
)
def test_a_declared_breakpoint_flag_picks_openais_dialect_only_where_the_request_goes_out_openai_shaped(
model: str, custom_llm_provider: str, expected: bool
) -> None:
assert (
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, custom_llm_provider, model_info={"supports_prompt_cache_breakpoint": True}
)
is expected
)
def test_a_gateway_deployment_declaring_the_flag_picks_openais_dialect_for_a_model_the_cost_map_lists_there() -> None:
litellm.register_model({"openrouter/qa-listed-gpt": {"litellm_provider": "openrouter", "mode": "chat"}})
assert [
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, "openrouter", model_info={"supports_prompt_cache_breakpoint": True}
)
for model in ("qa-listed-gpt", "qa-unlisted-gpt")
] == [True, False]
def test_a_gateway_deployment_declaring_the_flag_picks_openais_dialect_when_its_model_info_lists_that_provider() -> None:
assert [
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
"qa-self-hosted-gpt",
"hosted_vllm",
model_info={"supports_prompt_cache_breakpoint": True, "litellm_provider": listed_provider},
)
for listed_provider in ("hosted_vllm", "openrouter")
] == [True, False]
def test_a_custom_handler_provider_declaring_the_flag_keeps_the_anthropic_mark() -> None:
litellm.register_model(
{"qa-custom-handler/qa-model": {"litellm_provider": "qa-custom-handler", "mode": "chat"}}
)
assert (
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
"qa-model", "qa-custom-handler", model_info={"supports_prompt_cache_breakpoint": True}
)
is False
)
def test_a_routed_deployment_without_its_own_flag_reads_a_register_model_opt_in_but_not_a_siblings() -> None:
litellm.register_model(
{"azure_ai/qa-registered-gpt": {"litellm_provider": "azure_ai", "supports_prompt_cache_breakpoint": True}}
)
Router(
model_list=[
{
"model_name": "qa-opted-in-sibling",
"litellm_params": {"model": "azure_ai/qa-sibling-gpt", "api_base": "https://qa.services.ai.azure.com"},
"model_info": {"id": "qa-opted-in-sibling-id", "supports_prompt_cache_breakpoint": True},
}
]
)
assert [
AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model, "azure_ai", model_info={"id": "qa-flagless-id"}
)
for model in ("qa-registered-gpt", "qa-sibling-gpt")
] == [True, False]
@pytest.mark.parametrize(
("entrypoint", "injection_points"),
[
("completion", [{"location": "message", "role": "system"}]),
("acompletion", [{"location": "message", "role": "system"}]),
("aresponses", [{"location": "message", "role": "system"}]),
("aresponses-client-marked", None),
],
)
async def test_an_azure_custom_deployment_declaring_the_breakpoint_flag_gets_openais_breakpoint(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
entrypoint: str,
injection_points: list[dict[str, str]] | None,
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
azure: Final = respx_mock.post(host="qa-azure.openai.azure.com").mock(
side_effect=lambda request: Response(
200,
json=dict(
_FOUNDRY_RESPONSES_ROUTE[1] if request.url.path.endswith("/responses") else _FOUNDRY_CHAT_ROUTE[1]
),
)
)
deployment_params: Final = {
"model": "azure/my-custom-deployment",
"api_base": "https://qa-azure.openai.azure.com",
"api_key": "azure-test-key",
"api_version": "2025-04-01-preview",
}
router: Final = Router(
model_list=[
{
"model_name": "custom-azure",
"litellm_params": (
deployment_params
if injection_points is None
else {**deployment_params, "cache_control_injection_points": injection_points}
),
"model_info": {"id": "custom-azure-id", "supports_prompt_cache_breakpoint": True},
}
],
num_retries=0,
)
client_marked_system_prompt: Final = {
"role": "system",
"content": [{"type": "input_text", "text": "stable", "cache_control": {"type": "ephemeral"}}],
}
match entrypoint:
case "aresponses-client-marked":
await router.aresponses(
model="custom-azure", input=[client_marked_system_prompt, {"role": "user", "content": "hi"}]
)
case _:
await _send_unmarked_system_prompt(router, "custom-azure", entrypoint)
body: Final = azure.calls.last.request.content
assert (b"prompt_cache_breakpoint" in body, b"cache_control" in body) == (True, False)
_BEDROCK_CONVERSE_REPLY: Final = MappingProxyType(
{
"output": {"message": {"role": "assistant", "content": [{"text": "ok"}]}},
"stopReason": "end_turn",
"usage": {"inputTokens": 5, "outputTokens": 1, "totalTokens": 6},
}
)
_BEDROCK_INVOKE_REPLY: Final = MappingProxyType(
{
"id": "msg_bedrock",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-5-5",
"content": [{"type": "text", "text": "ok"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 5, "output_tokens": 1},
}
)
@pytest.mark.parametrize(
("entrypoint", "expected_system"),
[
("completion", [{"text": "stable"}, {"cachePoint": {"type": "default"}}]),
("acompletion", [{"text": "stable"}, {"cachePoint": {"type": "default"}}]),
("aanthropic_messages", [{"type": "text", "text": "stable", "cache_control": {"type": "ephemeral"}}]),
],
)
async def test_a_bedrock_claude_deployment_declaring_the_breakpoint_flag_keeps_bedrocks_own_cache_marks(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, entrypoint: str, expected_system: list[object]
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
bedrock: Final = respx_mock.post(host="bedrock-runtime.us-east-1.amazonaws.com").mock(
side_effect=lambda request: Response(
200,
json=dict(_BEDROCK_INVOKE_REPLY if request.url.path.endswith("/invoke") else _BEDROCK_CONVERSE_REPLY),
)
)
router: Final = Router(
model_list=[
{
"model_name": "claude-on-bedrock",
"litellm_params": {
"model": "bedrock/us.anthropic.claude-sonnet-5-5",
"aws_access_key_id": "bedrock-test-key",
"aws_secret_access_key": "bedrock-test-secret",
"aws_region_name": "us-east-1",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": "claude-on-bedrock-id", "supports_prompt_cache_breakpoint": True},
}
],
num_retries=0,
)
await _send_unmarked_system_prompt(router, "claude-on-bedrock", entrypoint)
body: Final = bedrock.calls.last.request.content
assert (json.loads(body)["system"], b"prompt_cache" in body) == (expected_system, False)
def _search_conversation_with(follow_up: dict[str, object]) -> list[AllMessageValues]:
return [
cast(AllMessageValues, {"role": "user", "content": "Weather in Paris?"}),

140
tests/unit/llms/anthropic/test_anthropic_common_utils.py
View file

@ -10,6 +10,7 @@ Verifies that:
- ANTHROPIC_API_KEY / ANTHROPIC_API_BASE take precedence over their aliases.
"""
import itertools
import json
import logging
import os
@ -4728,3 +4729,142 @@ def test_tools_without_strict_drops_only_the_strict_key_and_counts_the_strict_to
assert tools_without_strict(tools) == ToolsWithoutStrict(
tools=({"name": "a"}, {"name": "b"}, {"name": "c"}), dropped_count=1
)
@pytest.mark.parametrize("registration_order", [("caching-off", "unflagged"), ("unflagged", "caching-off")])
def test_auto_injection_reads_a_deployments_own_prompt_caching_flag_not_a_siblings(
registration_order: tuple[str, ...],
) -> None:
from litellm import Router
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
model: Final = "anthropic.claude-qa-cache-model"
litellm.register_model({model: {"litellm_provider": "bedrock", "mode": "chat", "supports_prompt_caching": True}})
model_infos: Final = {
"caching-off": {"id": "caching-off-id", "supports_prompt_caching": False},
"unflagged": {"id": "unflagged-id"},
}
Router(
model_list=[
{
"model_name": name,
"litellm_params": {"model": f"bedrock/{model}", "aws_region_name": "us-east-1"},
"model_info": model_infos[name],
}
for name in registration_order
]
)
assert {name: supports_anthropic_cache_control(model, "bedrock", model_infos[name]) for name in registration_order} == {
"caching-off": False,
"unflagged": True,
}
def test_auto_injection_leaves_an_unmapped_claude_deployment_uncached_instead_of_raising() -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
assert supports_anthropic_cache_control("claude-qa-unmapped-model", "anthropic", {"id": "unmapped-id"}) is False
def test_auto_injection_honors_a_register_model_opt_out_made_under_the_provider_prefixed_name() -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
from litellm.utils import catalog_with_runtime_registrations
model: Final = next(
key
for key, entry in catalog_with_runtime_registrations().items()
if entry.get("litellm_provider") == "anthropic" and entry.get("supports_prompt_caching") is True
)
litellm.register_model({f"anthropic/{model}": {"supports_prompt_caching": False}})
assert supports_anthropic_cache_control(model, "anthropic", {"id": "plain-id"}) is False
@pytest.mark.parametrize("route_prefix", ["bedrock/", "bedrock/converse/"])
def test_auto_injection_honors_a_register_model_opt_out_made_under_a_bedrock_route_prefixed_name(route_prefix: str) -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
from litellm.utils import catalog_with_runtime_registrations
model: Final = next(
key
for key, entry in catalog_with_runtime_registrations().items()
if entry.get("litellm_provider") == "bedrock_converse"
and entry.get("supports_prompt_caching") is True
and key.startswith("us.anthropic.claude")
)
litellm.register_model({f"{route_prefix}{model}": {"supports_prompt_caching": False}})
assert (
supports_anthropic_cache_control(model, "bedrock", {"id": "plain-id"}),
supports_anthropic_cache_control(model, "bedrock"),
) == (False, False)
@pytest.mark.parametrize(
("declared", "catalog_caching", "expected"),
[("true", False, True), (1, False, True), ("false", True, False), (0, True, False), ("maybe", True, True)],
)
def test_auto_injection_reads_a_string_or_numeric_deployment_prompt_caching_flag_as_the_bool_it_spells(
declared: object, catalog_caching: bool, expected: bool
) -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
model: Final = f"claude-qa-lax-flag-model-{catalog_caching}"
litellm.register_model(
{model: {"litellm_provider": "anthropic", "mode": "chat", "supports_prompt_caching": catalog_caching}}
)
assert (
supports_anthropic_cache_control(model, "anthropic", {"id": "lax-id", "supports_prompt_caching": declared})
is expected
)
@pytest.mark.parametrize("reload_cost_map", [False, True], ids=["before-reload", "after-reload"])
@pytest.mark.parametrize(
("registrations", "expected"),
[
((("", True), ("anthropic/", False)), False),
((("anthropic/", False), ("", True)), True),
((("", False), ("anthropic/", True), ("", False)), False),
],
ids=["prefixed-opt-out-last", "bare-opt-in-last", "bare-opt-out-registered-again-last"],
)
def test_the_later_of_a_bare_and_a_prefixed_register_model_entry_wins_for_routed_and_sdk_calls(
registrations: tuple[tuple[str, bool], ...], expected: bool, reload_cost_map: bool
) -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
from litellm.utils import catalog_with_runtime_registrations, reapply_runtime_model_cost_registrations
model: Final = next(
key
for key, entry in catalog_with_runtime_registrations().items()
if entry.get("litellm_provider") == "anthropic" and entry.get("supports_prompt_caching") is True
)
for prefix, flag in registrations:
litellm.register_model({f"{prefix}{model}": {"supports_prompt_caching": flag}})
if reload_cost_map:
reapply_runtime_model_cost_registrations()
assert (
supports_anthropic_cache_control(model, "anthropic", {"id": "plain-id"}),
supports_anthropic_cache_control(model, "anthropic"),
) == (expected, expected)
def test_a_register_model_opt_out_for_one_claude_model_leaves_every_other_routed_models_catalog_flag() -> None:
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
from litellm.utils import catalog_with_runtime_registrations
opted_out, untouched = itertools.islice(
(
key
for key, entry in catalog_with_runtime_registrations().items()
if entry.get("litellm_provider") == "anthropic" and entry.get("supports_prompt_caching") is True
),
2,
)
litellm.register_model({f"anthropic/{opted_out}": {"supports_prompt_caching": False}})
assert supports_anthropic_cache_control(untouched, "anthropic", {"id": "plain-id"}) is True

2
tests/unit/llms/azure/chat/test_azure_chat_gpt_transformation.py
View file

@ -917,7 +917,7 @@ def test_azure_sends_part_breakpoints_as_written_without_drop_params(
[
("unit-breakpoint-ignorer", {"drop_params": True}),
("unit-breakpoint-reader", {"drop_params": True, "base_model": "azure/unit-breakpoint-ignorer"}),
("my-azure-alias", {"drop_params": True, "model_info": {"base_model": "azure/unit-breakpoint-ignorer"}}),
("my-azure-alias", {"drop_params": True, "model_info": {"base_model": "azure/gpt-4.1"}}),
],
ids=("known-non-reader", "base-model-known-non-reader", "model-info-base-model-known-non-reader"),
)

184
tests/unit/llms/bedrock/test_bedrock_common_utils.py
View file

@ -1,5 +1,7 @@
import asyncio, importlib, litellm, pytest
import respx
from httpx import Response
from typing import Final
@ -581,6 +583,188 @@ def test_deployment_model_info_breakpoint_flag_covers_an_unmapped_arn(local_mode
assert bedrock_model_accepts_cache_points(unflagged_arn) is True
_CONVERSE_REPLY: Final = {
"output": {"message": {"role": "assistant", "content": [{"text": "ok"}]}},
"stopReason": "end_turn",
"usage": {"inputTokens": 5, "outputTokens": 1, "totalTokens": 6},
}
_CACHE_MARK: Final = {"type": "ephemeral"}
_MARKED_WEATHER_TOOL: Final = {
"type": "function",
"function": {"name": "get_weather", "parameters": {"type": "object", "properties": {"city": {"type": "string"}}}},
"cache_control": _CACHE_MARK,
}
_COMPUTER_TOOL: Final = {
"type": "computer_20250124",
"name": "computer",
"display_width_px": 1024,
"display_height_px": 768,
}
_MARKED_TOOL_CONVERSATION: Final = (
{"role": "system", "content": "stable"},
{"role": "user", "content": [{"type": "text", "text": "Weather in Paris and Rome?", "cache_control": _CACHE_MARK}]},
{
"role": "assistant",
"content": [{"type": "text", "text": "Checking both.", "cache_control": _CACHE_MARK}],
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": '{"city": "Paris"}{"city": "Rome"}'},
"cache_control": _CACHE_MARK,
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": [{"type": "text", "text": "Sunny", "cache_control": _CACHE_MARK}]},
{"role": "tool", "tool_call_id": "call_1_1", "content": "Cloudy"},
{"role": "assistant", "content": "Paris is sunny, Rome is cloudy.", "cache_control": _CACHE_MARK},
{"role": "user", "content": "Thanks"},
)
@pytest.mark.parametrize(
("model", "flagged", "declared", "tools"),
[
pytest.param("moonshotai.kimi-k3", "opted-in", True, (_MARKED_WEATHER_TOOL,), id="kimi-opted-in"),
pytest.param(
"us.anthropic.claude-sonnet-5-5",
"opted-out",
False,
(_COMPUTER_TOOL, _MARKED_WEATHER_TOOL),
id="claude-opted-out-next-to-computer-use",
),
],
)
@pytest.mark.parametrize("entrypoint", ["completion", "acompletion"])
@pytest.mark.parametrize("flagless_registers_first", [False, True])
async def test_a_bedrock_deployment_sends_cache_points_by_its_own_flag_not_a_siblings_on_the_same_model(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
local_model_cost_map: None,
model: str,
flagged: str,
declared: bool,
tools: tuple[dict[str, object], ...],
entrypoint: str,
flagless_registers_first: bool,
) -> None:
from litellm import Router
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
converse: Final = respx_mock.post(f"https://bedrock-runtime.us-east-1.amazonaws.com/model/{model}/converse").mock(
return_value=Response(200, json=_CONVERSE_REPLY)
)
flags: Final = {flagged: {"supports_prompt_cache_breakpoint": declared}}
registration_order: Final = ("flagless", flagged) if flagless_registers_first else (flagged, "flagless")
router: Final = Router(
model_list=[
{
"model_name": name,
"litellm_params": {
"model": f"bedrock/{model}",
"aws_access_key_id": "bedrock-test-key",
"aws_secret_access_key": "bedrock-test-secret",
"aws_region_name": "us-east-1",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": f"{name}-id", **flags.get(name, {})},
}
for name in registration_order
],
num_retries=0,
)
messages: Final = [dict(message) for message in _MARKED_TOOL_CONVERSATION]
async def sends_a_cache_point(name: str) -> bool:
match entrypoint:
case "completion":
router.completion(model=name, messages=messages, tools=list(tools))
case "acompletion":
await router.acompletion(model=name, messages=messages, tools=list(tools))
return b"cachePoint" in converse.calls.last.request.content
assert {name: await sends_a_cache_point(name) for name in registration_order} == {
flagged: declared,
"flagless": not declared,
}
async def test_a_routed_bedrock_deployment_sends_cache_points_when_register_model_opted_its_model_in(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, local_model_cost_map: None
) -> None:
from litellm import Router
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
converse: Final = respx_mock.post(
"https://bedrock-runtime.us-east-1.amazonaws.com/model/moonshotai.kimi-k3/converse"
).mock(return_value=Response(200, json=_CONVERSE_REPLY))
litellm.register_model({"bedrock/moonshotai.kimi-k3": {"supports_prompt_cache_breakpoint": True}})
router: Final = Router(
model_list=[
{
"model_name": "flagless",
"litellm_params": {
"model": "bedrock/moonshotai.kimi-k3",
"aws_access_key_id": "bedrock-test-key",
"aws_secret_access_key": "bedrock-test-secret",
"aws_region_name": "us-east-1",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": "flagless-id"},
}
],
num_retries=0,
)
await router.acompletion(model="flagless", messages=[{"role": "system", "content": "stable"}, {"role": "user", "content": "hi"}])
assert b"cachePoint" in converse.calls.last.request.content
@pytest.mark.parametrize(
("catalog_caching", "deployment_caching", "expected"),
[(None, True, True), (True, False, False), (True, None, True), (None, None, False)],
)
def test_a_routed_deployments_own_prompt_caching_flag_decides_cache_points_for_a_catalogued_model(
catalog_caching: bool | None, deployment_caching: bool | None, expected: bool, local_model_cost_map: None
) -> None:
from litellm.llms.bedrock.common_utils import bedrock_model_accepts_cache_points
model: Final = f"qa-catalogued-model-{catalog_caching}-{deployment_caching}"
catalog_flag: Final = {} if catalog_caching is None else {"supports_prompt_caching": catalog_caching}
deployment_flag: Final = {} if deployment_caching is None else {"supports_prompt_caching": deployment_caching}
litellm.register_model({model: {"litellm_provider": "bedrock", "mode": "chat", **catalog_flag}})
assert bedrock_model_accepts_cache_points(model, model_info={"id": "qa-deployment", **deployment_flag}) is expected
@pytest.mark.parametrize(
("declared", "catalog_caching", "expected"),
[
("true", False, True),
(1, False, True),
(" TRUE ", False, True),
("false", True, False),
(0, True, False),
("maybe", True, True),
("maybe", False, False),
],
)
def test_a_deployments_prompt_caching_flag_written_as_a_string_or_number_decides_cache_points_as_the_bool_it_spells(
declared: object, catalog_caching: bool, expected: bool, local_model_cost_map: None
) -> None:
from litellm.llms.bedrock.common_utils import bedrock_model_accepts_cache_points
model: Final = f"qa-catalogued-model-lax-{catalog_caching}"
litellm.register_model(
{model: {"litellm_provider": "bedrock", "mode": "chat", "supports_prompt_caching": catalog_caching}}
)
assert (
bedrock_model_accepts_cache_points(model, model_info={"id": "qa-deployment", "supports_prompt_caching": declared})
is expected
)
def test_merge_bedrock_aws_request_params_strips_caller_identity_when_deployment_has_static_credentials():
from litellm.llms.bedrock.common_utils import merge_bedrock_aws_request_params

42
tests/unit/llms/openai/chat/test_openai_gpt_transformation.py
View file

@ -2173,3 +2173,45 @@ async def test_openai_request_rewrites_media_parts_without_touching_the_callers_
{"type": "file", "file": {"file_data": "data:application/pdf;base64,JVBERi0=", "filename": "my_file.pdf"}},
]
assert messages == messages_before
_OPENAI_DEPLOYMENT_FLAGS: Final = {"opted-in": {"supports_prompt_cache_breakpoint": True}}
@pytest.mark.parametrize("registration_order", [("opted-in", "flagless"), ("flagless", "opted-in")])
async def test_chat_sends_a_breakpoint_by_the_deployments_own_flag_not_a_siblings_on_the_same_model(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, registration_order: tuple[str, str]
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
catalog_row_without_a_flag: Final = {"litellm_provider": "openai", "mode": "chat"}
monkeypatch.setitem(litellm.model_cost, "qa-openai-shared", catalog_row_without_a_flag)
chat: Final = respx_mock.post("https://api.openai.com/v1/chat/completions").mock(
return_value=Response(200, json=_OPENAI_CHAT_REPLY)
)
router: Final = litellm.Router(
model_list=[
{
"model_name": name,
"litellm_params": {
"model": "openai/qa-openai-shared",
"api_base": "https://api.openai.com/v1",
"api_key": "openai-test-key",
"cache_control_injection_points": [{"location": "message", "role": "system"}],
},
"model_info": {"id": f"{name}-id", **_OPENAI_DEPLOYMENT_FLAGS.get(name, {})},
}
for name in registration_order
],
num_retries=0,
)
async def sends_a_breakpoint(name: str) -> bool:
await router.acompletion(
model=name, messages=[{"role": "system", "content": "stable"}, {"role": "user", "content": "hi"}]
)
return b"prompt_cache_breakpoint" in chat.calls.last.request.content
assert {name: await sends_a_breakpoint(name) for name in registration_order} == {
"opted-in": True,
"flagless": False,
}

25
tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py
View file

@ -6564,6 +6564,31 @@ def test_an_application_inference_profile_gets_no_item_marks_under_drop_params()
assert AnthropicCacheControlHook.count_request_cache_breakpoints(messages) == 0
def test_a_bedrock_deployment_keeps_item_marks_by_its_own_flag_not_a_siblings_on_the_same_model(
local_model_cost_map: None,
):
from litellm import Router
Router(
model_list=[
{
"model_name": "opted-in",
"litellm_params": {"model": "bedrock/moonshotai.kimi-k3", "aws_region_name": "us-east-1"},
"model_info": {"id": "opted-in-id", "supports_prompt_cache_breakpoint": True},
}
]
)
marked_system: Final = {"role": "system", "content": "House rules", "cache_control": _EPHEMERAL}
def bridged_system(model_info: Mapping[str, object]) -> Mapping[str, object]:
return _bridged_request(marked_system, model="bedrock/moonshotai.kimi-k3", model_info=model_info)[1]
assert (
bridged_system({"id": "flagless-id"}),
bridged_system({"id": "opted-in-id", "supports_prompt_cache_breakpoint": True}),
) == ({"role": "system", "content": "House rules"}, marked_system)
def test_the_shared_input_transform_keeps_every_item_mark_for_a_request_it_sends():
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=cast(ResponseInputParam, list(_marked_turns(5))), # cast-ok: cache_control is not an input TypedDict key

20
tests/unit/test_utils.py
View file

@ -11433,6 +11433,26 @@ def test_vertex_llama3_model_info_resolves_without_the_meta_prefix():
assert prefixed_info["key"] == f"vertex_ai/{model}", model
def test_the_catalog_with_register_model_entries_lists_each_model_once_and_leaves_router_registrations_out() -> None:
from litellm.utils import catalog_with_runtime_registrations
catalog: Final = catalog_with_runtime_registrations()
catalog_size: Final = len(catalog)
catalogued_model: Final = next(iter(catalog))
catalogued_entry: Final = dict(catalog[catalogued_model])
litellm.register_model({catalogued_model: {"supports_prompt_cache_breakpoint": True}})
litellm.register_model({"qa-registered-only-model": {"litellm_provider": "openai", "mode": "chat"}})
litellm.register_model({"qa-router-deployment-id": {"mode": "chat"}}, persist_across_reloads=False)
merged: Final = catalog_with_runtime_registrations()
listed: Final = list(merged)
assert len(listed) == len(merged) == catalog_size + 1
assert "qa-registered-only-model" in listed
assert "qa-router-deployment-id" not in listed
assert dict(merged[catalogued_model]) == {**catalogued_entry, "supports_prompt_cache_breakpoint": True}
@pytest.mark.parametrize(
"key", ("openai.test-converse", "bedrock/openai.test-converse", "bedrock/converse/openai.test-converse")
)