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

fix(edenai): restore unrelated files the squashed PR commit had reverted

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yassin 2026-09-21 19:42:37 +00:00
parent 889b8fb220
commit 594d97e7e7
11 changed files with 844 additions and 251 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1115,7 +1115,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
responses_tools: Final[list[ALL_RESPONSES_API_TOOL_PARAMS]] = []
for tool in tools:
# convert function tool from chat completion to responses API format
if tool.get("type") == "function":
if tool.get("type") == "function" and isinstance(tool.get("function"), dict):
function_tool = cast(ChatCompletionToolParamFunctionChunk, tool.get("function"))
responses_tools.append(
FunctionToolParam(

320
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -36,6 +36,7 @@ from litellm.types.integrations.anthropic_cache_control_hook import (
CacheControlMessageInjectionPoint,
)
from litellm.types.llms.anthropic import (
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES,
AllAnthropicToolsValues,
AnthropicSystemMessageContent,
)
@ -124,6 +125,16 @@ def _carries_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and any(block.get(key) is not None for key in CACHE_BREAKPOINT_KEYS)
def _tool_carries_cache_breakpoint(tool: object) -> bool:
return _carries_cache_breakpoint(tool) or (
isinstance(tool, dict) and _carries_cache_breakpoint(tool.get("function"))
)
def _chat_transform_drops_tool_cache_control(tool: object) -> bool:
return isinstance(tool, dict) and tool.get("type") in ANTHROPIC_TOOL_SEARCH_TOOL_TYPES
def _accepts_prompt_cache_breakpoint(block: object) -> bool:
return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES
@ -134,6 +145,8 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool:
# rather than spending them on a list that is still missing some of their targets.
CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_points"
EXTERNAL_BREAKPOINTS_STAMP: Final = "_litellm_external_breakpoints"
class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
@ -199,19 +212,13 @@ class AnthropicCacheControlHook(CustomPromptManagement):
# Create a deep copy of messages to avoid modifying the original list
processed_messages = copy.deepcopy(messages)
# Separate message-level and non-message-level injection points
message_points: Final[list[CacheControlMessageInjectionPoint]] = []
remaining_points: Final[list[CacheControlInjectionPoint]] = []
for point in injection_points:
if point.get("location") == "message":
message_points.append(cast(CacheControlMessageInjectionPoint, point))
else:
remaining_points.append(point)
message_points: Final = tuple(
cast(CacheControlMessageInjectionPoint, point)
for point in injection_points
if point.get("location") == "message"
)
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
# Non-message points (currently Bedrock tool_config) are handled in the
# provider transform, where each tool_config point appends at most one
# cachePoint to the tools. That block also counts toward Anthropic's
# limit, so reserve a slot for it here to leave room.
stamped_dialect: Final = injection_points[0].get("_litellm_openai_dialect")
openai_dialect: Final = (
stamped_dialect
@ -236,8 +243,10 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if carry_unmatched
else tuple(message_points)
)
reserved_blocks: Final = (
1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0
stamped_external: Final = injection_points[0].get(EXTERNAL_BREAKPOINTS_STAMP)
external_breakpoints: Final = stamped_external if isinstance(stamped_external, int) else 0
reserved_blocks: Final = AnthropicCacheControlHook._blocks_reserved_outside_messages(
remaining_points, external_breakpoints, openai_dialect
)
breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages)
processed_messages = self._apply_message_injections(
@ -254,14 +263,19 @@ class AnthropicCacheControlHook(CustomPromptManagement):
# Points this pass did not place: non-message ones for the provider transform, and
# the deferred role-targeted ones. Deferring is what reaches the Responses API's
# `instructions`, which is only a system message once the bridge builds one. The
# judged stamp is what makes it safe: the next pass must not re-judge points
# against messages this pass already marked (see `_should_stand_down`).
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (*remaining_points, *carried_message_points)
# `instructions`, which is only a system message once the bridge builds one. A later
# pass re-applies them safely: a target that already carries a mark is skipped and
# the census counts every mark on the wire, litellm's own included.
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (
*AnthropicCacheControlHook._points_with_a_slot_left(
remaining_points,
AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages) + external_breakpoints,
openai_dialect,
),
*carried_message_points,
)
if carried_points:
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged(
carried_points
)
non_default_params["cache_control_injection_points"] = list(carried_points)
return model, processed_messages, non_default_params
@ -296,6 +310,72 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
return system_blocks + sum(AnthropicCacheControlHook._count_cache_control_blocks(msg) for msg in messages)
@staticmethod
def count_external_cache_breakpoints(
tools: Iterable[object] | None, cache_control: object = None, request_kwargs: object = None
) -> int:
"""Client breakpoints outside messages and system that the provider cap still counts.
A tool carries its mark at the top level (Anthropic shape) or under ``function``
(OpenAI shape). A top-level ``cache_control`` is Anthropic's automatic caching,
which places one breakpoint of its own on top of the explicit ones. The
``extra_body`` envelope of ``request_kwargs`` is merged over the request on the
wire, so a ``tools`` or ``cache_control`` it carries replaces the direct value
and is counted in its place. Callers pass only the tools whose mark reaches the
provider on their path.
"""
extra_body: Final = (
_validated_object_mapping(AnthropicCacheControlHook._request_value(request_kwargs, "extra_body")) or {}
)
wire_cache_control: Final = extra_body.get("cache_control", cache_control)
wire_tools: Final = _validated_object_list(extra_body["tools"]) if "tools" in extra_body else tools
tool_blocks: Final = sum(1 for tool in wire_tools or () if _tool_carries_cache_breakpoint(tool))
envelope_blocks: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(
_validated_object_list(extra_body.get("messages")) or (), extra_body.get("system")
)
return int(wire_cache_control is not None) + tool_blocks + envelope_blocks
@staticmethod
def count_external_cache_breakpoints_on_messages_route(
tools: Iterable[object] | None, cache_control: object, request_kwargs: object
) -> int:
"""The /v1/messages census before the route splits.
The native messages transforms drop the ``extra_body`` envelope while the
chat bridge merges it, so the cap reserves for whichever census is larger
rather than letting an envelope that unmarks a direct tool free a slot the
provider still counts.
"""
return max(
AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control),
AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control, request_kwargs),
)
@staticmethod
def _blocks_reserved_outside_messages(
remaining_points: Sequence[CacheControlInjectionPoint], external_breakpoints: int, openai_dialect: bool
) -> int:
"""Slots of the provider cap that the message census cannot see.
The client's breakpoints on tools and its automatic top-level ``cache_control``
are already on the wire, and a ``tool_config`` point becomes one more cachePoint
in the Bedrock converse transform. OpenAI's cap counts only its own block markers.
"""
if openai_dialect:
return 0
tool_config_blocks: Final = 1 if any(p.get("location") == "tool_config" for p in remaining_points) else 0
return external_breakpoints + tool_config_blocks
@staticmethod
def _points_with_a_slot_left(
remaining_points: Sequence[CacheControlInjectionPoint], breakpoints_on_wire: int, openai_dialect: bool
) -> tuple[CacheControlInjectionPoint, ...]:
"""A ``tool_config`` point becomes a cachePoint the Bedrock converse transform never
counts against the cap, so it is forwarded only while the wire still has a slot."""
if openai_dialect or breakpoints_on_wire < MAX_CACHE_CONTROL_BLOCKS:
return tuple(remaining_points)
return tuple(point for point in remaining_points if point.get("location") != "tool_config")
@staticmethod
def _apply_message_injections(
points: Sequence[CacheControlMessageInjectionPoint],
@ -476,11 +556,16 @@ class AnthropicCacheControlHook(CustomPromptManagement):
def apply_to_anthropic_messages_request(
messages: list[dict],
system: str | list | None,
injection_points: list[CacheControlInjectionPoint],
injection_points: Sequence[CacheControlInjectionPoint],
openai_dialect: bool = False,
external_breakpoints: int = 0,
) -> tuple[list[dict], str | list | None, list[CacheControlInjectionPoint]]:
"""Apply cache control injection for the Anthropic-native v1/messages endpoint.
``external_breakpoints`` is the client's breakpoint count outside ``messages`` and
``system`` (see ``count_external_cache_breakpoints``); it shrinks the budget so
the request never exceeds the provider cap.
Returns (messages, system, remaining_non_message_points).
"""
if not injection_points:
@ -489,22 +574,17 @@ class AnthropicCacheControlHook(CustomPromptManagement):
processed_messages: list[dict] = copy.deepcopy(messages)
processed_system = copy.deepcopy(system) if system is not None else None
message_points: Final[list[CacheControlMessageInjectionPoint]] = []
system_points: Final[list[CacheControlMessageInjectionPoint]] = []
remaining_points: Final[list[CacheControlInjectionPoint]] = []
role_points: Final = tuple(
cast(CacheControlMessageInjectionPoint, point)
for point in injection_points
if point.get("location") == "message"
)
system_points: Final = tuple(point for point in role_points if point.get("role") == "system")
message_points: Final = tuple(point for point in role_points if point.get("role") != "system")
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
for point in injection_points:
if point.get("location") == "message":
msg_point = cast(CacheControlMessageInjectionPoint, point)
if msg_point.get("role") == "system":
system_points.append(msg_point)
else:
message_points.append(msg_point)
else:
remaining_points.append(point)
reserved_blocks: Final = (
1 if not openai_dialect and any(p.get("location") == "tool_config" for p in remaining_points) else 0
reserved_blocks: Final = AnthropicCacheControlHook._blocks_reserved_outside_messages(
remaining_points, external_breakpoints, openai_dialect
)
max_blocks: Final = MAX_CACHE_CONTROL_BLOCKS - reserved_blocks
@ -541,8 +621,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
max_blocks=max_blocks - system_blocks,
openai_dialect=openai_dialect,
)
forwarded_points: Final = AnthropicCacheControlHook._points_with_a_slot_left(
remaining_points,
AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages, processed_system)
+ external_breakpoints,
openai_dialect,
)
return processed_messages, processed_system, remaining_points
return processed_messages, processed_system, list(forwarded_points)
@staticmethod
def _default_control() -> ChatCompletionCachedContent:
@ -559,31 +645,26 @@ class AnthropicCacheControlHook(CustomPromptManagement):
return ChatCompletionCachedContent(type="ephemeral")
@staticmethod
def _stamped_as_judged(points: Sequence[CacheControlInjectionPoint]) -> Sequence[Mapping[str, object]]:
"""Mark written-back points as having passed the client cache_control judgment.
Builds copies because config-owned point dicts are shared across
requests; mutating them would leak the stamp into future requests.
"""
return AnthropicCacheControlHook._stamped(points, "_litellm_judged", True)
@staticmethod
def _judged_configured_points(
def _stamped_for_prompt_hook(
points: Sequence[CacheControlInjectionPoint],
messages: list[AllMessageValues],
tools: list[object] | None,
cache_control: object,
external_breakpoints: int,
model: str,
custom_llm_provider: str | None,
api_base: object,
prompt_cache_options: object,
request_kwargs: object,
) -> Sequence[Mapping[str, object]] | None:
if AnthropicCacheControlHook._should_stand_down(points, messages, None, tools, cache_control, request_kwargs):
return None
return AnthropicCacheControlHook._stamped_with_dialect(
) -> Sequence[Mapping[str, object]]:
"""Carry onto the points what the prompt-management hook never receives.
The hook sees neither the tools nor the request kwargs, so the target dialect
and the client's breakpoint count outside the message list ride on the points.
Builds copies because config-owned point dicts are shared across requests.
"""
with_dialect: Final = AnthropicCacheControlHook._stamped_with_dialect(
points, model, custom_llm_provider, api_base, prompt_cache_options
)
if external_breakpoints == 0:
return with_dialect
return AnthropicCacheControlHook._stamped(with_dialect, EXTERNAL_BREAKPOINTS_STAMP, external_breakpoints)
@staticmethod
def _stamped_with_dialect(
@ -604,35 +685,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
@staticmethod
def _stamped(
points: Sequence[CacheControlInjectionPoint], key: str, value: object
) -> Sequence[Mapping[str, object]]:
def _stamped(points: Sequence[Mapping[str, object]], key: str, value: object) -> Sequence[Mapping[str, object]]:
return [{**point, key: value} for point in points]
@staticmethod
def _should_stand_down(
points: Sequence[CacheControlInjectionPoint],
messages: list[AllMessageValues],
system: str | list | None,
tools: list | None,
cache_control: object = None,
request_kwargs: object = None,
) -> bool:
"""Whether configured injection points must yield to client-set cache_control.
Points that a prior pass over this request already judged and wrote
back carry the internal judged stamp; any re-entry (acompletion
re-entering completion, the async-to-sync /v1/messages dispatch,
interceptor sub-calls reusing the request kwargs) must not re-judge
them, because by then the messages carry litellm's own injected marks
and the judgment would misread those as client breakpoints.
"""
if all(point.get("_litellm_judged") for point in points):
return False
return AnthropicCacheControlHook._request_has_cache_control(
messages, system, tools, cache_control, request_kwargs
)
@staticmethod
def _request_has_cache_control(
messages: list[AllMessageValues],
@ -641,27 +696,18 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control: object = None,
request_kwargs: object = None,
) -> bool:
"""Client breakpoints own caching in both the request and its extra_body envelope."""
bodies: Final = (
{"messages": messages, "system": system, "tools": tools, "cache_control": cache_control},
_validated_object_mapping(AnthropicCacheControlHook._request_value(request_kwargs, "extra_body")) or {},
)
return any(
body.get("cache_control") is not None
or AnthropicCacheControlHook.count_request_cache_breakpoints(
_validated_object_list(body.get("messages")) or (), body.get("system")
)
> 0
or any(
AnthropicCacheControlHook._request_value(tool, "cache_control") is not None
or AnthropicCacheControlHook._request_value(
AnthropicCacheControlHook._request_value(tool, "function"), "cache_control"
)
is not None
for tool in (_validated_object_list(body.get("tools")) or ())
)
for body in bodies
)
"""Return True if the request already carries any client-supplied cache_control.
Only the automatic defaults stand down on it: a client that marks its own
breakpoints (Claude Code does) has a caching strategy the defaults would
clash with, whether the marks sit in the request or in its ``extra_body``
envelope. Configured injection points are an explicit instruction and are
applied alongside the client's marks, bounded by the provider cap.
"""
return (
AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system)
+ AnthropicCacheControlHook.count_external_cache_breakpoints(tools, cache_control, request_kwargs)
) > 0
@staticmethod
def get_default_injection_points(
@ -769,34 +815,30 @@ class AnthropicCacheControlHook(CustomPromptManagement):
) -> None:
"""For /chat/completions: resolve the injection points the request should carry.
Configured injection points win over the automatic defaults, but stand
down entirely when the client already marked its own cache_control
breakpoints (messages or tools): injecting alongside them clashes with
the client's caching strategy and can exceed the provider's four-block
limit. The judgment happens once per request; points a prior pass
wrote back carry the judged stamp and are never re-judged (see
``_should_stand_down``). Seeding the param lets the existing
prompt-management gate and the AnthropicCacheControlHook run
unchanged.
Configured injection points win over the automatic defaults and are applied
even when the client marked its own cache_control elsewhere in the request;
the provider's four-block cap bounds them, counting the client's marks on
messages, tools and the top-level ``cache_control``. Only the defaults stand
down on client marks. Seeding the param lets the existing prompt-management
gate and the AnthropicCacheControlHook run unchanged.
"""
import litellm
if non_default_params.get("cache_control_injection_points"):
judged: Final = AnthropicCacheControlHook._judged_configured_points(
non_default_params["cache_control_injection_points"],
messages,
tools,
non_default_params.get("cache_control"),
configured: Final = non_default_params.get("cache_control_injection_points")
if configured:
tools_keeping_marks: Final = tuple(
tool for tool in tools or () if not _chat_transform_drops_tool_cache_control(tool)
)
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_for_prompt_hook(
configured,
AnthropicCacheControlHook.count_external_cache_breakpoints(
tools_keeping_marks, non_default_params.get("cache_control"), non_default_params
),
model,
custom_llm_provider,
api_base,
non_default_params.get("prompt_cache_options"),
non_default_params,
)
if judged is None:
non_default_params.pop("cache_control_injection_points")
else:
non_default_params["cache_control_injection_points"] = judged
return
points: Final = AnthropicCacheControlHook.get_default_injection_points(
messages=messages,
@ -897,15 +939,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
) -> tuple[list[dict], str | list | None]:
"""Extract cache_control_injection_points from kwargs and apply if present.
Configured points stand down entirely when the client already marked
its own cache_control breakpoints anywhere in the request. The
judgment happens once per request; points a prior pass wrote back
carry the judged stamp and are never re-judged (see
``_should_stand_down``). When none are configured but
Configured points are applied even when the client marked its own
cache_control elsewhere in the request, bounded by the provider cap,
which counts the client's marks on messages, system, tools and the
top-level ``cache_control``. When none are configured but
``litellm.enable_anthropic_prompt_caching`` or the per-request
``enable_prompt_caching`` kwarg (stamped from key metadata) is on,
synthesize default breakpoints for the native /v1/messages path. Pops
both keys from kwargs;
synthesize default breakpoints for the native /v1/messages path; those
defaults alone stand down on client marks. Pops both keys from kwargs;
if remaining (non-message) points exist they are written back so
downstream transforms can handle them.
"""
@ -917,13 +958,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
configured: Final = cast( # cast-ok: kwargs is untyped; this key only holds the documented injection-point list
list[CacheControlInjectionPoint] | None, kwargs.pop("cache_control_injection_points", None)
)
if configured and AnthropicCacheControlHook._should_stand_down(
configured, typed_messages, system, tools, cache_control, kwargs
):
return messages, system
injection_points: list[CacheControlInjectionPoint] = configured or []
if not injection_points and model is not None:
injection_points = AnthropicCacheControlHook.get_default_injection_points(
injection_points: Final[Sequence[CacheControlInjectionPoint]] = configured or (
AnthropicCacheControlHook.get_default_injection_points(
messages=typed_messages,
system=system,
tools=tools,
@ -933,6 +969,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
cache_control=cache_control,
request_kwargs=kwargs,
)
if model is not None
else ()
)
if not injection_points:
return messages, system
@ -945,6 +984,9 @@ class AnthropicCacheControlHook(CustomPromptManagement):
system=system,
injection_points=injection_points,
openai_dialect=openai_dialect,
external_breakpoints=AnthropicCacheControlHook.count_external_cache_breakpoints_on_messages_route(
tools, cache_control, kwargs
),
)
breakpoints_added: Final = (
AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system) - breakpoints_before
@ -953,7 +995,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if openai_dialect and breakpoints_added > 0:
kwargs.setdefault("prompt_cache_options", PromptCacheOptions(mode="explicit"))
if remaining:
kwargs["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_as_judged(remaining)
kwargs["cache_control_injection_points"] = remaining
return messages, system
@property

9
litellm/llms/azure_ai/common_utils.py
View file

@ -6,6 +6,7 @@ from urllib.parse import urlparse
import litellm
from litellm.llms.base_llm.base_utils import BaseLLMModelInfo, BaseTokenCounter
from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.openai import AllMessageValues
from litellm.types.router import GenericLiteLLMParams
@ -150,6 +151,14 @@ def azure_ai_supports_native_responses(model: str | None, api_base: str | None)
return AzureFoundryModelInfo.get_azure_ai_route(model) == "default"
def foundry_chat_rejects_function_tools_while_reasoning(
model: str, reasoning_effort: str | Mapping[str, object] | None
) -> bool:
if reasoning_effort is None:
return OpenAIGPT5Config.is_model_gpt_6_plus_model(model)
return OpenAIGPT5Config.is_model_gpt_5_6_plus_model(model)
class AzureFoundryModelInfo(BaseLLMModelInfo):
"""Model info for Azure AI / Azure Foundry models."""

33
litellm/llms/openai/chat/gpt_5_transformation.py
View file

@ -1,5 +1,6 @@
"""Support for OpenAI gpt-5 model family."""
import re
from typing import Final
import litellm
@ -11,6 +12,8 @@ from litellm.utils import (
from .gpt_transformation import OpenAIGPTConfig
_GPT_SERIES_VERSION: Final = re.compile(r"^gpt-(\d+)(?:\.(\d+))?(?=[.-]|$)")
def _catalogue_declares_default_effort() -> bool:
"""Whether the loaded cost map carries default_reasoning_effort for ANY entry.
@ -112,20 +115,28 @@ class OpenAIGPT5Config(OpenAIGPTConfig):
model_name: Final = model.split("/")[-1]
return model_name.startswith("gpt-5.4")
@staticmethod
def _gpt_series_version(model: str) -> tuple[int, int] | None:
match: Final = _GPT_SERIES_VERSION.match(model.split("/")[-1])
if match is None:
return None
return int(match.group(1)), int(match.group(2) or 0)
@classmethod
def is_model_gpt_5_4_plus_model(cls, model: str) -> bool:
"""Check if the model is gpt-5.4 or newer (5.4, 5.5, 5.6, etc., including pro)."""
model_name: Final = model.split("/")[-1]
if model_name.startswith("gpt-6"):
return True
if not model_name.startswith("gpt-5."):
return False
try:
version_str: Final = model_name.replace("gpt-5.", "").split("-")[0]
major: Final = version_str.split(".")[0]
return int(major) >= 4
except (ValueError, IndexError):
return False
version: Final = cls._gpt_series_version(model)
return version is not None and version >= (5, 4)
@classmethod
def is_model_gpt_5_6_plus_model(cls, model: str) -> bool:
version: Final = cls._gpt_series_version(model)
return version is not None and version >= (5, 6)
@classmethod
def is_model_gpt_6_plus_model(cls, model: str) -> bool:
version: Final = cls._gpt_series_version(model)
return version is not None and version >= (6, 0)
@classmethod
def _model_map_lookup_name(cls, model: str) -> str:

47
litellm/main.py
View file

@ -100,6 +100,10 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
from litellm.litellm_core_utils.request_timeout_resolver import (
get_configured_request_timeout,
)
from litellm.llms.azure_ai.common_utils import (
azure_ai_supports_native_responses,
foundry_chat_rejects_function_tools_while_reasoning,
)
from litellm.llms.base_llm import BaseConfig, BaseImageGenerationConfig
from litellm.llms.base_llm.base_model_iterator import (
convert_model_response_to_streaming,
@ -1106,10 +1110,18 @@ def responses_api_bridge_check(
# provider with a custom api_base and gpt-5.4+ model names serve tools without
# reasoning fine and have no /responses route, so they keep pre-existing
# behavior (bridge only on an explicit reasoning_effort).
# - Azure AI Foundry's OpenAI v1 hosts (azure_ai provider) enforce it later in the series:
# an explicit effort with function tools is rejected from gpt-5.6 on, and the unset
# effort only from gpt-6 on (gpt-5.6 serves tools with reasoning silently off), so the
# azure_ai gate keys on those measured boundaries instead of gpt-5.4+.
# - Older GPT-5 names (e.g. ``gpt-5``, ``gpt-5.1``): bridge only when a reasoning
# summary alias is present with ``reasoning_effort`` (tools alone stay on chat).
has_function_tool: Final = any(
(tool.get("type") == "function" if isinstance(tool, dict) else getattr(tool, "type", None) == "function")
(
tool.get("type") == "function" and (isinstance(tool.get("function"), dict) or "name" in tool)
if isinstance(tool, dict)
else getattr(tool, "type", None) == "function"
)
for tool in (tools or ())
)
if isinstance(reasoning_effort, dict):
@ -1118,28 +1130,35 @@ def responses_api_bridge_check(
reasoning_active = reasoning_effort != "none"
# The reasoning+tools constraint is enforced by the real OpenAI backend behind any api.openai.com
# host (the default URL or a PrivateLink hostname such as <region>.privatelink.api.openai.com) and
# by Azure OpenAI. Resolve the effective base arg>global>env>default exactly as the chat handler
# does, so a custom base set via litellm.api_base or OPENAI_BASE_URL/OPENAI_API_BASE isn't misread
# as the default and bridged to a /responses route it lacks. A whitespace-only base collapses to
# the default too.
# by Azure OpenAI through the azure provider. Resolve the effective OpenAI base arg>global>env>default
# exactly as the chat handler does, so a custom base set via litellm.api_base or
# OPENAI_BASE_URL/OPENAI_API_BASE isn't misread as the default and bridged to a /responses route it
# lacks. A whitespace-only base collapses to the default too.
resolved_api_base: Final = _resolve_openai_api_base(api_base).strip()
on_foundry_openai_endpoint: Final = custom_llm_provider == "azure_ai" and azure_ai_supports_native_responses(
model, api_base
)
on_constraint_enforcing_endpoint: Final = (
custom_llm_provider == "azure" or resolved_api_base == "" or _is_openai_backed_api_base(resolved_api_base)
)
if (
custom_llm_provider in ("openai", "azure")
and model_info.get("mode") != "responses"
and OpenAIGPT5Config.is_model_gpt_5_model(model)
and not OpenAIGPT5Config.is_model_gpt_5_search_model(model)
chat_rejects_function_tools: Final = (
has_function_tool
and reasoning_active
and (
(reasoning_effort is not None and reasoning_summary is not None)
or (
foundry_chat_rejects_function_tools_while_reasoning(model, reasoning_effort)
if on_foundry_openai_endpoint
else (
OpenAIGPT5Config.is_model_gpt_5_4_plus_model(model)
and has_function_tool
and reasoning_active
and (reasoning_effort is not None or on_constraint_enforcing_endpoint)
)
)
)
if (
(custom_llm_provider in ("openai", "azure") or on_foundry_openai_endpoint)
and model_info.get("mode") != "responses"
and OpenAIGPT5Config.is_model_gpt_5_model(model)
and not OpenAIGPT5Config.is_model_gpt_5_search_model(model)
and ((reasoning_effort is not None and reasoning_summary is not None) or chat_rejects_function_tools)
):
model_info["mode"] = "responses"
model = model.replace("responses/", "")

4
litellm/types/integrations/anthropic_cache_control_hook.py
View file

@ -17,8 +17,8 @@ class CacheControlMessageInjectionPoint(TypedDict):
role: Literal["user", "system", "assistant"] | None # Optional: target by role (user, system, assistant)
index: int | str | None # Optional: target by specific index
control: ChatCompletionCachedContent | None
_litellm_judged: NotRequired[bool] # Internal: written back by litellm once the client cache_control judgment ran
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
class CacheControlToolConfigInjectionPoint(TypedDict):
@ -26,8 +26,8 @@ class CacheControlToolConfigInjectionPoint(TypedDict):
location: Literal["tool_config"]
control: ChatCompletionCachedContent | None
_litellm_judged: NotRequired[bool] # Internal: written back by litellm once the client cache_control judgment ran
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
CacheControlInjectionPoint = CacheControlMessageInjectionPoint | CacheControlToolConfigInjectionPoint

4
litellm/types/llms/anthropic.py
View file

@ -756,6 +756,10 @@ class ANTHROPIC_BETA_HEADER_VALUES(str, Enum):
# Tool search beta header constant (for Anthropic direct API and Microsoft Foundry)
ANTHROPIC_TOOL_SEARCH_BETA_HEADER: Final = "advanced-tool-use-2025-11-20"
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES: Final = frozenset(
{"tool_search_tool_regex_20251119", "tool_search_tool_bm25_20251119"}
)
# Effort beta header constant
ANTHROPIC_EFFORT_BETA_HEADER: Final = "effort-2025-11-24"

18
tests/test_litellm/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
View file

@ -830,6 +830,24 @@ def test_convert_tools_to_responses_format():
assert result[0]["name"] == "test"
def test_convert_tools_to_responses_format_passes_flat_function_tool_through():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
LiteLLMResponsesTransformationHandler,
)
handler = LiteLLMResponsesTransformationHandler()
flat_tool = {
"type": "function",
"name": "shell",
"description": "Run a shell command",
"parameters": {"type": "object", "properties": {"cmd": {"type": "string"}}, "required": ["cmd"]},
}
converted = handler._convert_tools_to_responses_format([flat_tool])
assert converted == [flat_tool]
def test_extract_extra_body_params_reasoning_effort_override():
"""Test that reasoning_effort from extra_body overrides top-level reasoning_effort"""
from litellm.completion_extras.litellm_responses_transformation.transformation import (

440
tests/test_litellm/integrations/test_anthropic_cache_control_hook.py
View file

@ -4,10 +4,11 @@ import os
import subprocess
import sys
import textwrap
from typing import List, Optional, Tuple
from typing import Final, List, Optional, Tuple
from unittest.mock import MagicMock, patch
import pytest
from pydantic import BaseModel, ConfigDict
import litellm
from litellm.integrations.anthropic_cache_control_hook import (
@ -1276,11 +1277,7 @@ def test_cache_control_hook_reserves_slot_for_tool_config_point():
)
assert _count_cache_control(processed) == 3
# The tool_config point is passed through for the provider transform,
# stamped so re-entries never re-judge it against litellm's own marks.
assert non_default_params["cache_control_injection_points"] == [
{"location": "tool_config", "_litellm_judged": True}
]
assert non_default_params["cache_control_injection_points"] == [{"location": "tool_config"}]
@pytest.mark.asyncio
@ -1338,18 +1335,8 @@ async def test_cache_control_hook_bedrock_payload_caps_with_tool_config_point(mo
client=client,
)
request_body = json.loads(mock_post.call_args.kwargs["data"])
cache_points = sum(
1 for block in request_body.get("system", []) if isinstance(block, dict) and "cachePoint" in block
)
for msg in request_body.get("messages", []):
content = msg.get("content", [])
if isinstance(content, list):
cache_points += sum(1 for block in content if isinstance(block, dict) and "cachePoint" in block)
for tool in request_body.get("toolConfig", {}).get("tools", []):
if isinstance(tool, dict) and "cachePoint" in tool:
cache_points += 1
request_body = _ConverseBody.model_validate_json(mock_post.call_args.kwargs["data"])
cache_points = _count_converse_cache_points(request_body)
assert cache_points <= 4, (
f"Bedrock payload exceeded Anthropic's 4 cache_control block limit "
@ -1357,6 +1344,97 @@ async def test_cache_control_hook_bedrock_payload_caps_with_tool_config_point(mo
)
class _ConverseMessage(BaseModel):
model_config = ConfigDict(frozen=True)
content: tuple[dict[str, object], ...] = ()
class _ConverseToolConfig(BaseModel):
model_config = ConfigDict(frozen=True)
tools: tuple[dict[str, object], ...] = ()
class _ConverseBody(BaseModel):
model_config = ConfigDict(frozen=True)
system: tuple[dict[str, object], ...] = ()
messages: tuple[_ConverseMessage, ...] = ()
toolConfig: _ConverseToolConfig = _ConverseToolConfig()
def _count_converse_cache_points(request_body: _ConverseBody) -> int:
blocks: Final = (
*request_body.system,
*(block for message in request_body.messages for block in message.content),
*request_body.toolConfig.tools,
)
return sum(1 for block in blocks if "cachePoint" in block)
@pytest.mark.asyncio
async def test_cache_control_hook_bedrock_tool_config_point_stands_down_when_client_marks_fill_the_cap(
monkeypatch: pytest.MonkeyPatch,
):
with patch.dict(
os.environ,
{
"AWS_ACCESS_KEY_ID": "fake_access_key_id",
"AWS_SECRET_ACCESS_KEY": "fake_secret_access_key",
"AWS_REGION_NAME": "us-east-1",
},
):
monkeypatch.setattr(litellm, "callbacks", [AnthropicCacheControlHook()])
mock_response = MagicMock()
mock_response.json.return_value = {
"output": {"message": {"role": "assistant", "content": "ok"}},
"stopReason": "end_turn",
"usage": {"inputTokens": 100, "outputTokens": 4, "totalTokens": 104},
}
mock_response.status_code = 200
client = AsyncHTTPHandler()
with patch.object(client, "post", return_value=mock_response) as mock_post:
marked = {"type": "ephemeral"}
messages = [
{"role": "system", "content": [{"type": "text", "text": "sys", "cache_control": marked}]},
*(
{"role": "user", "content": [{"type": "text", "text": f"turn {i}", "cache_control": marked}]}
for i in range(3)
),
{"role": "user", "content": "What is the weather?"},
]
await litellm.acompletion(
model="bedrock/us.anthropic.claude-opus-4-6-v1:0",
messages=messages,
max_tokens=32,
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
"required": ["location"],
},
},
}
],
cache_control_injection_points=[{"location": "tool_config"}],
client=client,
)
request_body = _ConverseBody.model_validate_json(mock_post.call_args.kwargs["data"])
assert _count_converse_cache_points(request_body) == 4
assert not any("cachePoint" in tool for tool in request_body.toolConfig.tools)
class TestApplyToAnthropicMessagesRequest:
"""Tests for apply_to_anthropic_messages_request (v1/messages cache control)."""
@ -1683,13 +1761,17 @@ class TestEnableAnthropicPromptCaching:
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
messages, system, kwargs, model, provider, tools=tools,
)
if client_control != "none":
if client_control != "none" and not configured:
assert (result_messages, result_system, tools) == original
assert kwargs["metadata"] == {}
else:
assert kwargs["metadata"]["litellm_gateway_injected_cache"] == "selected-deployment"
assert sum(AnthropicCacheControlHook._count_cache_control_blocks(m) for m in result_messages) == 1
assert result_system[0]["cache_control"] == control
assert result_messages[-1]["content"][-1]["cache_control"] == control
assert tools == original[2]
assert (result_messages == original[0]) == (envelope == "request" and client_control == "message")
assert (result_system == original[1]) == (envelope == "request" and client_control == "system")
if provider == "vertex_ai":
wire = VertexAIAnthropicConfig().transform_request(
model=model, messages=[{"role": "system", "content": result_system}, *result_messages],
@ -1706,7 +1788,7 @@ class TestEnableAnthropicPromptCaching:
AnthropicCacheControlHook.maybe_seed_default_injection_points(
seeded, [{"role": "system", "content": original[1]}, *original[0]], model, provider, tools=tools,
)
assert bool(seeded.get("cache_control_injection_points")) == (client_control == "none")
assert bool(seeded.get("cache_control_injection_points")) == (client_control == "none" or configured)
@pytest.mark.asyncio
@pytest.mark.parametrize("asynchronous", [False, True])
@ -2257,13 +2339,11 @@ class TestPerKeyEnablePromptCaching:
assert result_msgs == messages
class TestConfiguredInjectionPointsStandDown:
"""Configured cache_control_injection_points must stand down entirely when the
client already set its own cache_control anywhere in the request (LIT-4582);
injecting alongside client breakpoints clashes with the client's caching
strategy and can push the request past Anthropic's four-block limit."""
class TestConfiguredInjectionPointsSurviveClientMarks:
CONFIGURED = [{"location": "message", "role": "system"}]
TAIL_POINT = [{"location": "message", "index": -1}]
TOOL_CONFIG_POINT = [{"location": "tool_config"}]
EPHEMERAL = {"type": "ephemeral"}
CLEAN_MESSAGES: List[AllMessageValues] = [
{"role": "system", "content": "sys"},
@ -2277,6 +2357,37 @@ class TestConfiguredInjectionPointsStandDown:
V1_MESSAGES = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
MARKED_TOOL_TOP_LEVEL = {
"type": "function",
"function": {"name": "t", "parameters": {}},
"cache_control": {"type": "ephemeral"},
}
MARKED_TOOL_NESTED = {
"type": "function",
"function": {"name": "t", "parameters": {}, "cache_control": {"type": "ephemeral"}},
}
UNMARKED_TOOL = {"type": "function", "function": {"name": "t", "parameters": {}}}
MARKED_V1_TOOL = {"name": "t", "input_schema": {}, "cache_control": {"type": "ephemeral"}}
UNMARKED_V1_TOOL = {"name": "t", "input_schema": {}}
MARKED_SYSTEM = [{"type": "text", "text": "sys", "cache_control": EPHEMERAL}]
MARKED_TOOL_SEARCH_REGEX = {
"type": "tool_search_tool_regex_20251119",
"name": "tool_search",
"cache_control": {"type": "ephemeral"},
}
MARKED_TOOL_SEARCH_BM25 = {
"type": "tool_search_tool_bm25_20251119",
"name": "tool_search",
"cache_control": {"type": "ephemeral"},
}
@staticmethod
def _marked_user_turns(count: int) -> List[AllMessageValues]:
return [
{"role": "user", "content": [{"type": "text", "text": f"turn {i}", "cache_control": {"type": "ephemeral"}}]}
for i in range(count)
]
def _seed(self, params, messages, tools=None):
AnthropicCacheControlHook.maybe_seed_default_injection_points(
non_default_params=params,
@ -2286,6 +2397,17 @@ class TestConfiguredInjectionPointsStandDown:
tools=tools,
)
def _chat(self, params: dict[str, object], messages: List[AllMessageValues]) -> List[AllMessageValues]:
_, processed, _ = AnthropicCacheControlHook().get_chat_completion_prompt(
model="claude-sonnet-4-5",
messages=messages,
non_default_params=params,
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
return processed
def _inject(self, messages, kwargs, system="sys", tools=None):
return AnthropicCacheControlHook.maybe_inject_cache_control(
messages,
@ -2296,23 +2418,79 @@ class TestConfiguredInjectionPointsStandDown:
tools=tools,
)
def test_configured_points_dropped_when_messages_carry_cache_control(self):
def test_chat_tail_point_applies_when_client_marked_the_system_block(self):
messages: List[AllMessageValues] = [
{"role": "system", "content": [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]},
{"role": "user", "content": "history"},
{"role": "assistant", "content": "reply"},
{"role": "user", "content": "question"},
]
params = {"cache_control_injection_points": copy.deepcopy(self.TAIL_POINT)}
self._seed(params, messages)
processed = self._chat(params, messages)
assert processed[0] == messages[0]
assert processed[-1] == {"role": "user", "content": "question", "cache_control": self.EPHEMERAL}
assert _count_cache_control(processed) == 2
def test_chat_configured_points_apply_when_messages_carry_cache_control(self):
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(self.MARKED_MESSAGES))
assert "cache_control_injection_points" not in params
processed = self._chat(params, copy.deepcopy(self.MARKED_MESSAGES))
assert processed[0] == {"role": "system", "content": "sys", "cache_control": self.EPHEMERAL}
assert processed[1] == self.MARKED_MESSAGES[1]
@pytest.mark.parametrize(
"tool",
[
{"type": "function", "function": {"name": "t", "parameters": {}}, "cache_control": {"type": "ephemeral"}},
{"type": "function", "function": {"name": "t", "parameters": {}, "cache_control": {"type": "ephemeral"}}},
],
ids=["top_level", "nested_in_function"],
"tool", [MARKED_TOOL_TOP_LEVEL, MARKED_TOOL_NESTED], ids=["top_level", "nested_in_function"]
)
def test_configured_points_dropped_when_tools_carry_cache_control(self, tool):
def test_chat_configured_points_apply_when_tools_carry_cache_control(self, tool):
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(self.CLEAN_MESSAGES), tools=[tool])
assert "cache_control_injection_points" not in params
processed = self._chat(params, copy.deepcopy(self.CLEAN_MESSAGES))
assert processed[0] == {"role": "system", "content": "sys", "cache_control": self.EPHEMERAL}
@pytest.mark.parametrize(
"tool,injected",
[(MARKED_TOOL_TOP_LEVEL, 0), (MARKED_TOOL_NESTED, 0), (UNMARKED_TOOL, 1)],
ids=["marked_top_level", "marked_nested_in_function", "unmarked"],
)
def test_chat_cap_counts_client_marked_tools(self, tool, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(messages), tools=[tool])
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 3 + injected
@pytest.mark.parametrize("tool", [MARKED_TOOL_SEARCH_REGEX, MARKED_TOOL_SEARCH_BM25], ids=["regex", "bm25"])
def test_chat_cap_ignores_marked_tool_search_tools(self, tool):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
self._seed(params, copy.deepcopy(messages), tools=[tool])
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 4
@pytest.mark.parametrize("marked_turns,forwarded", [(3, ["tool_config"]), (4, [])], ids=["slot_left", "cap_full"])
def test_chat_forwards_tool_config_point_only_while_a_slot_is_left(self, marked_turns, forwarded):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
params = {"cache_control_injection_points": copy.deepcopy(self.TOOL_CONFIG_POINT)}
self._seed(params, copy.deepcopy(messages), tools=[self.UNMARKED_TOOL])
self._chat(params, copy.deepcopy(messages))
assert [p["location"] for p in params.get("cache_control_injection_points", [])] == forwarded
@pytest.mark.parametrize("marked_turns,forwarded", [(3, ["tool_config"]), (4, [])], ids=["slot_left", "cap_full"])
def test_v1_messages_forwards_tool_config_point_only_while_a_slot_is_left(self, marked_turns, forwarded):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.TOOL_CONFIG_POINT)}
self._inject(self._marked_user_turns(marked_turns), kwargs, tools=[self.UNMARKED_V1_TOOL])
assert [p["location"] for p in kwargs.get("cache_control_injection_points", [])] == forwarded
@pytest.mark.parametrize("marked_turns,injected", [(2, 1), (3, 0)])
def test_chat_root_cache_control_reserves_a_slot(self, marked_turns, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
root_cache_control = {"type": "ephemeral"}
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "cache_control": root_cache_control}
self._seed(params, copy.deepcopy(messages))
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == marked_turns + injected
assert params["cache_control"] is root_cache_control
def test_configured_points_kept_when_request_is_unmarked(self):
configured = copy.deepcopy(self.CONFIGURED)
@ -2320,43 +2498,59 @@ class TestConfiguredInjectionPointsStandDown:
self._seed(params, copy.deepcopy(self.CLEAN_MESSAGES))
assert params["cache_control_injection_points"] is configured
def test_judged_remainder_survives_reentry_despite_injected_marks(self):
"""acompletion() re-enters completion() after injection ran, with only the
stamped non-message points written back; the re-entry must not misread
litellm's own marks as client ones and drop that remainder."""
remainder = [{"location": "tool_config", "_litellm_judged": True}]
params = {"cache_control_injection_points": remainder}
self._seed(params, copy.deepcopy(self.MARKED_MESSAGES))
assert params["cache_control_injection_points"] is remainder
def test_chat_reentry_over_injected_messages_adds_no_duplicate_marks(self):
points = [{"location": "message", "role": "system"}, {"location": "tool_config"}]
first_params = {"cache_control_injection_points": copy.deepcopy(points)}
self._seed(first_params, copy.deepcopy(self.MARKED_MESSAGES))
first = self._chat(first_params, copy.deepcopy(self.MARKED_MESSAGES))
assert _count_cache_control(first) == 2
assert first_params["cache_control_injection_points"] == [{"location": "tool_config"}]
def test_v1_messages_stand_down_when_content_block_marked(self):
second_params = {"cache_control_injection_points": copy.deepcopy(points)}
self._seed(second_params, copy.deepcopy(first))
second = self._chat(second_params, copy.deepcopy(first))
assert second == first
assert second_params["cache_control_injection_points"] == [{"location": "tool_config"}]
def test_v1_messages_configured_point_applies_when_content_block_marked(self):
messages = [
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": {"type": "ephemeral"}}]}
]
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
result_msgs, result_sys = self._inject(copy.deepcopy(messages), kwargs)
assert result_msgs == messages
assert result_sys == "sys"
assert result_sys == [{"type": "text", "text": "sys", "cache_control": self.EPHEMERAL}]
assert "cache_control_injection_points" not in kwargs
def test_v1_messages_stand_down_when_system_block_marked(self):
"""A configured point targeting a message must not fire when the client
marked the system prompt; the old behavior injected into the message
because only the exact targeted position was guarded."""
def test_v1_messages_tail_point_applies_when_system_block_marked(self):
system = [{"type": "text", "text": "s", "cache_control": {"type": "ephemeral"}}]
kwargs = {"cache_control_injection_points": [{"location": "message", "role": "user"}]}
kwargs = {"cache_control_injection_points": copy.deepcopy(self.TAIL_POINT)}
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, system=system)
assert result_msgs == self.V1_MESSAGES
assert result_msgs == [
{"role": "user", "content": [{"type": "text", "text": "hi", "cache_control": self.EPHEMERAL}]}
]
assert result_sys == system
assert "cache_control_injection_points" not in kwargs
def test_v1_messages_stand_down_when_tools_marked(self):
tools = [{"name": "t", "input_schema": {}, "cache_control": {"type": "ephemeral"}}]
def test_v1_messages_configured_point_applies_when_tools_marked(self):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, tools=tools)
result_msgs, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs, tools=[self.MARKED_V1_TOOL])
assert result_msgs == self.V1_MESSAGES
assert result_sys == "sys"
assert "cache_control_injection_points" not in kwargs
assert result_sys == [{"type": "text", "text": "sys", "cache_control": self.EPHEMERAL}]
@pytest.mark.parametrize(
"tool,expected_system",
[
(MARKED_V1_TOOL, "sys"),
(MARKED_TOOL_SEARCH_REGEX, "sys"),
(MARKED_TOOL_SEARCH_BM25, "sys"),
(UNMARKED_V1_TOOL, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]),
],
ids=["marked", "marked_tool_search_regex", "marked_tool_search_bm25", "unmarked"],
)
def test_v1_messages_cap_counts_client_marked_tools(self, tool, expected_system):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
_, result_sys = self._inject(self._marked_user_turns(3), kwargs, tools=[tool])
assert result_sys == expected_system
def test_v1_messages_configured_points_apply_when_unmarked(self):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED)}
@ -2364,16 +2558,73 @@ class TestConfiguredInjectionPointsStandDown:
assert result_sys == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]
@pytest.mark.parametrize(
"configured",
[None, CONFIGURED],
ids=["automatic_defaults", "configured_points"],
"extra_body,injected",
[
({"tools": [MARKED_TOOL_TOP_LEVEL]}, 0),
({"cache_control": {"type": "ephemeral"}}, 0),
({"tools": [UNMARKED_TOOL]}, 1),
],
ids=["marked_tool", "root_cache_control", "unmarked_tool"],
)
def test_v1_messages_stands_down_for_root_cache_control(self, monkeypatch, configured):
def test_chat_cap_counts_client_marks_sent_through_extra_body(self, extra_body, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(3)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "extra_body": extra_body}
self._seed(params, copy.deepcopy(messages))
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == 3 + injected
@pytest.mark.parametrize(
"extra_body,expected_system",
[
({"cache_control": {"type": "ephemeral"}}, "sys"),
({"tools": [MARKED_V1_TOOL]}, "sys"),
({"tools": [UNMARKED_V1_TOOL]}, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]),
],
ids=["root_cache_control", "marked_tool", "unmarked_tool"],
)
def test_v1_messages_cap_counts_client_marks_sent_through_extra_body(self, extra_body, expected_system):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), "extra_body": extra_body}
_, result_sys = self._inject(self._marked_user_turns(3), kwargs)
assert result_sys == expected_system
@pytest.mark.parametrize(
"params,tools,marked_turns,injected",
[
({"extra_body": {"tools": [MARKED_TOOL_TOP_LEVEL]}}, [MARKED_TOOL_TOP_LEVEL], 2, 1),
({"extra_body": {"tools": [UNMARKED_TOOL]}}, [MARKED_TOOL_TOP_LEVEL], 3, 1),
({"extra_body": {"tools": [MARKED_TOOL_TOP_LEVEL]}}, [UNMARKED_TOOL], 3, 0),
({"extra_body": {"cache_control": EPHEMERAL}, "cache_control": EPHEMERAL}, None, 2, 1),
],
ids=["same_marked_tool_both_ways", "extra_body_unmarks", "extra_body_marks", "root_cache_control_both_ways"],
)
def test_chat_cap_counts_extra_body_fields_in_place_of_the_direct_ones(self, params, tools, marked_turns, injected):
messages = [{"role": "system", "content": "sys"}, *self._marked_user_turns(marked_turns)]
params = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), **copy.deepcopy(params)}
self._seed(params, copy.deepcopy(messages), tools=tools)
processed = self._chat(params, copy.deepcopy(messages))
assert _count_cache_control(processed) == marked_turns + injected
@pytest.mark.parametrize(
"kwargs,tools,marked_turns,expected_system",
[
({"extra_body": {"tools": [MARKED_V1_TOOL]}}, [MARKED_V1_TOOL], 2, MARKED_SYSTEM),
({"extra_body": {"tools": [UNMARKED_V1_TOOL]}}, [MARKED_V1_TOOL], 3, "sys"),
({"extra_body": {"tools": [MARKED_V1_TOOL]}}, [UNMARKED_V1_TOOL], 3, "sys"),
({"extra_body": {"cache_control": EPHEMERAL}, "cache_control": EPHEMERAL}, None, 2, MARKED_SYSTEM),
],
ids=["same_marked_tool_both_ways", "extra_body_unmarks", "extra_body_marks", "root_cache_control_both_ways"],
)
def test_v1_messages_cap_reserves_for_the_larger_of_direct_and_extra_body_marks(
self, kwargs, tools, marked_turns, expected_system
):
kwargs = {"cache_control_injection_points": copy.deepcopy(self.CONFIGURED), **copy.deepcopy(kwargs)}
_, result_sys = self._inject(self._marked_user_turns(marked_turns), kwargs, tools=tools)
assert result_sys == expected_system
def test_v1_messages_automatic_defaults_stand_down_for_root_cache_control(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
root_cache_control = {"type": "ephemeral"}
kwargs = {"cache_control": root_cache_control, "litellm_metadata": {}}
if configured is not None:
kwargs["cache_control_injection_points"] = copy.deepcopy(configured)
result_messages, result_system = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
@ -2382,17 +2633,28 @@ class TestConfiguredInjectionPointsStandDown:
assert kwargs["cache_control"] is root_cache_control
assert "litellm_gateway_injected_cache" not in kwargs["litellm_metadata"]
@pytest.mark.parametrize(
"marked_turns,expected_system",
[(2, [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]), (3, "sys")],
)
def test_v1_messages_configured_points_apply_with_root_cache_control_reserving_a_slot(
self, marked_turns, expected_system
):
root_cache_control = {"type": "ephemeral"}
kwargs = {
"cache_control": root_cache_control,
"cache_control_injection_points": copy.deepcopy(self.CONFIGURED),
}
_, result_system = self._inject(self._marked_user_turns(marked_turns), kwargs)
assert result_system == expected_system
assert kwargs["cache_control"] is root_cache_control
def test_v1_messages_reentry_flow_preserves_tool_config_remainder(self):
"""The advisor interceptor re-enters anthropic_messages() with the outer
request's kwargs and post-injection messages. The first pass applies the
message point and writes back a stamped tool_config remainder; the
re-entry must keep that remainder even though the messages and system
now carry litellm's own marks."""
points = [{"location": "message", "role": "system"}, {"location": "tool_config"}]
kwargs = {"cache_control_injection_points": copy.deepcopy(points)}
msgs1, sys1 = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
assert sys1[0]["cache_control"] == {"type": "ephemeral"}
expected_remainder = [{"location": "tool_config", "_litellm_judged": True}]
expected_remainder = [{"location": "tool_config"}]
assert kwargs["cache_control_injection_points"] == expected_remainder
msgs2, sys2 = self._inject(msgs1, kwargs, system=sys1)
@ -2631,22 +2893,26 @@ class TestOpenAIPromptCacheBreakpoint:
assert system == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]
assert kwargs == {}
def test_v1_messages_client_content_breakpoint_makes_configured_points_stand_down(self):
messages = [{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}]
def test_v1_messages_configured_points_apply_beside_client_content_breakpoint(self):
messages = [
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}
]
kwargs = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
result, system = self._inject(messages, "sys", kwargs)
assert result == messages
assert system == "sys"
assert kwargs == {}
assert system == [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
assert kwargs == {"prompt_cache_options": self.EXPLICIT}
def test_v1_messages_client_system_breakpoint_makes_configured_points_stand_down(self):
def test_v1_messages_tail_point_applies_beside_client_system_breakpoint(self):
system = [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
messages = [{"role": "user", "content": [{"type": "text", "text": "hi"}]}]
kwargs = {"cache_control_injection_points": [{"location": "message", "index": -1}]}
result, result_system = self._inject(messages, system, kwargs)
assert result == messages
assert result == [
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]}
]
assert result_system == system
assert kwargs == {}
assert kwargs == {"prompt_cache_options": self.EXPLICIT}
def test_chat_system_string_wrapped_with_block_breakpoint(self):
params = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
@ -2710,18 +2976,25 @@ class TestOpenAIPromptCacheBreakpoint:
assert processed[0] == {"role": "system", "content": "sys", "cache_control": {"type": "ephemeral"}}
assert params == {}
def test_chat_client_breakpoint_makes_seeded_points_stand_down(self):
def test_chat_seeded_points_apply_beside_client_breakpoint(self):
params = {"cache_control_injection_points": copy.deepcopy(self.SYSTEM_POINT)}
messages = [
{"role": "system", "content": "sys"},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]},
]
AnthropicCacheControlHook.maybe_seed_default_injection_points(
non_default_params=params,
messages=[
{"role": "system", "content": "sys"},
{"role": "user", "content": [{"type": "text", "text": "hi", "prompt_cache_breakpoint": self.EXPLICIT}]},
],
messages=messages,
model="openai/gpt-5.6",
custom_llm_provider="openai",
)
assert params == {}
assert params["cache_control_injection_points"] == [
{"location": "message", "role": "system", "_litellm_openai_dialect": True}
]
_, processed, _ = self._chat(messages, params)
assert processed[0]["content"] == [{"type": "text", "text": "sys", "prompt_cache_breakpoint": self.EXPLICIT}]
assert processed[1] == messages[1]
assert params["prompt_cache_options"] == self.EXPLICIT
def test_cap_counts_client_breakpoints_of_both_kinds(self):
messages = [
@ -3315,7 +3588,6 @@ class TestRecordGatewayInjection:
assert kwargs["litellm_metadata"][self.KEY] == self.DEPLOYMENT
def test_configured_points_skipping_a_marked_target_record_nothing(self):
"""Configured injection stands down on client breakpoints, so no marker lands."""
kwargs: dict = {
"litellm_metadata": {},
"cache_control_injection_points": [{"location": "message", "role": "system", "index": None}],

52
tests/test_litellm/llms/openai/test_is_model_gpt_5_model.py
View file

@ -159,6 +159,58 @@ class TestOpenAIGPT5ConfigIsModelGpt54PlusModel:
), f"Expected '{model}' NOT to be classified as gpt-5.4-or-newer"
GPT5_6_PLUS_MODELS = [
"gpt-6-astra",
"openai/gpt-6-astra",
"gpt-5.6",
"gpt-5.6-sol",
"gpt-5.6-terra",
"gpt-5.10-preview",
]
GPT5_PRE_5_6_MODELS = [
"gpt-5",
"gpt-5.4",
"gpt-5.4-mini",
"gpt-5.5",
"gpt-5.5-pro",
"gpt-4o",
]
GPT6_PLUS_MODELS = [
"gpt-6-astra",
"openai/gpt-6-astra",
"gpt-6",
"gpt-6.1-preview",
]
GPT_PRE_6_MODELS = [
"gpt-5.6-sol",
"gpt-5.5",
"gpt-5",
"gpt-4o",
]
class TestOpenAIGPT5ConfigSeriesBoundaries:
@pytest.mark.parametrize("model", GPT5_6_PLUS_MODELS)
def test_gpt5_6_plus_models_are_classified_as_5_6_plus(self, model: str):
assert OpenAIGPT5Config.is_model_gpt_5_6_plus_model(model)
@pytest.mark.parametrize("model", GPT5_PRE_5_6_MODELS)
def test_pre_5_6_models_are_not_classified_as_5_6_plus(self, model: str):
assert not OpenAIGPT5Config.is_model_gpt_5_6_plus_model(model)
@pytest.mark.parametrize("model", GPT6_PLUS_MODELS)
def test_gpt6_plus_models_are_classified_as_6_plus(self, model: str):
assert OpenAIGPT5Config.is_model_gpt_6_plus_model(model)
@pytest.mark.parametrize("model", GPT_PRE_6_MODELS)
def test_pre_6_models_are_not_classified_as_6_plus(self, model: str):
assert not OpenAIGPT5Config.is_model_gpt_6_plus_model(model)
# ---------------------------------------------------------------------------
# AzureOpenAIGPT5Config
# ---------------------------------------------------------------------------

166
tests/test_litellm/test_main.py
View file

@ -1049,6 +1049,35 @@ def test_responses_api_bridge_check_gpt_5_4_flat_function_tool_routes_to_respons
assert model_info.get("mode") == "responses"
@pytest.mark.parametrize(
"custom_llm_provider, model_name, api_base",
[
pytest.param("openai", "gpt-5.6", None, id="openai"),
pytest.param("azure_ai", "gpt-6-astra", "https://myproject.services.ai.azure.com", id="azure-ai-foundry"),
],
)
def test_responses_api_bridge_check_function_tool_without_body_stays_chat(
monkeypatch, custom_llm_provider, model_name, api_base
):
import litellm
from litellm.main import responses_api_bridge_check
monkeypatch.delenv("OPENAI_BASE_URL", raising=False)
monkeypatch.delenv("OPENAI_API_BASE", raising=False)
monkeypatch.setattr(litellm, "api_base", None)
model_info, model = responses_api_bridge_check(
model=model_name,
custom_llm_provider=custom_llm_provider,
tools=[{"type": "function"}],
reasoning_effort=None,
api_base=api_base,
)
assert model == model_name
assert model_info.get("mode") != "responses"
def test_responses_api_bridge_check_dict_effort_none_stays_chat():
"""The escape hatch must honor litellm's dict form: {"effort": "none"} means reasoning off."""
from litellm.main import responses_api_bridge_check
@ -1308,6 +1337,68 @@ def test_responses_api_bridge_check_azure_with_api_base_and_unset_effort_routes(
assert model_info.get("mode") == "responses"
_FOUNDRY_API_BASE: Final = "https://myproject.services.ai.azure.com"
_FOUNDRY_FUNCTION_TOOL: Final = ({"type": "function", "function": {"name": "get_weather"}},)
@pytest.mark.parametrize(
"model_name, api_base, reasoning_effort",
[
pytest.param("gpt-6-astra", _FOUNDRY_API_BASE, None, id="gpt-6-unset-effort"),
pytest.param("gpt-6-astra", _FOUNDRY_API_BASE, "low", id="gpt-6-explicit-effort"),
pytest.param("gpt-6-astra", "https://myresource.openai.azure.com", None, id="gpt-6-azure-openai-host"),
pytest.param("gpt-5.6-sol", _FOUNDRY_API_BASE, "low", id="gpt-5.6-explicit-effort"),
pytest.param("gpt-5.6-sol", _FOUNDRY_API_BASE, {"effort": "high"}, id="gpt-5.6-explicit-effort-dict"),
],
)
def test_responses_api_bridge_check_azure_ai_foundry_rejected_tools_route_to_responses(
model_name, api_base, reasoning_effort
):
from litellm.main import responses_api_bridge_check
model_info, model = responses_api_bridge_check(
model=model_name,
custom_llm_provider="azure_ai",
tools=_FOUNDRY_FUNCTION_TOOL,
reasoning_effort=reasoning_effort,
api_base=api_base,
)
assert model == model_name
assert model_info.get("mode") == "responses"
@pytest.mark.parametrize(
"model_name, api_base, reasoning_effort",
[
pytest.param("gpt-6-astra", _FOUNDRY_API_BASE, "none", id="explicit-none-stays-chat"),
pytest.param("gpt-5.6-sol", _FOUNDRY_API_BASE, None, id="gpt-5.6-unset-effort-stays-chat"),
pytest.param("gpt-5.6-sol", _FOUNDRY_API_BASE, "none", id="gpt-5.6-explicit-none-stays-chat"),
pytest.param("gpt-5.5", _FOUNDRY_API_BASE, "high", id="gpt-5.5-explicit-effort-stays-chat"),
pytest.param("gpt-5.4-mini", _FOUNDRY_API_BASE, None, id="gpt-5.4-mini-unset-effort-stays-chat"),
pytest.param("gpt-5.4-mini", _FOUNDRY_API_BASE, "low", id="gpt-5.4-mini-explicit-effort-stays-chat"),
pytest.param("gpt-6-astra", "https://myproject.models.ai.azure.com", None, id="serverless-host-stays-chat"),
pytest.param("Mistral-large-2411", _FOUNDRY_API_BASE, None, id="non-gpt-5-model-stays-chat"),
pytest.param("claude-opus-4-1", _FOUNDRY_API_BASE, None, id="claude-on-foundry-stays-chat"),
],
)
def test_responses_api_bridge_check_azure_ai_without_foundry_responses_route_stays_chat(
model_name, api_base, reasoning_effort
):
from litellm.main import responses_api_bridge_check
model_info, model = responses_api_bridge_check(
model=model_name,
custom_llm_provider="azure_ai",
tools=_FOUNDRY_FUNCTION_TOOL,
reasoning_effort=reasoning_effort,
api_base=api_base,
)
assert model == model_name
assert model_info.get("mode") != "responses"
def test_responses_api_bridge_check_older_gpt_5_tools_without_reasoning_stays_chat():
"""Pre-5.4 GPT-5 names keep the old boundary: tools alone never bridge."""
from litellm.main import responses_api_bridge_check
@ -1488,6 +1579,81 @@ def test_responses_bridge_preserves_reasoning_effort_with_drop_params(
assert request_body["reasoning"] == {"effort": "high"}
_FOUNDRY_RESPONSES_FUNCTION_CALL_BODY: Final = {
"id": "resp_foundry",
"object": "response",
"created_at": 1789852145,
"status": "completed",
"model": "gpt-6-astra",
"output": [
{
"id": "fc_1",
"type": "function_call",
"status": "completed",
"arguments": '{"city":"Paris"}',
"call_id": "call_1",
"name": "get_weather",
}
],
"parallel_tool_calls": True,
"usage": {
"input_tokens": 53,
"output_tokens": 18,
"total_tokens": 71,
"output_tokens_details": {"reasoning_tokens": 0},
},
"error": None,
"incomplete_details": None,
"instructions": None,
"metadata": {},
"temperature": 1.0,
"tool_choice": "auto",
"tools": [],
"top_p": 1.0,
"max_output_tokens": 200,
"previous_response_id": None,
"reasoning": {"effort": "medium", "summary": None},
"truncation": "disabled",
"user": None,
}
def test_completion_bridges_azure_ai_foundry_gpt_5_4_plus_function_tools_to_responses(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
):
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
responses_route: Final = respx_mock.post(f"{_FOUNDRY_API_BASE}/openai/v1/responses").respond(
json=_FOUNDRY_RESPONSES_FUNCTION_CALL_BODY
)
response: Final = litellm.completion(
model="azure_ai/gpt-6-astra",
messages=[{"role": "user", "content": "What is the weather in Paris? Use the tool."}],
tools=[
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
},
}
],
max_tokens=200,
api_base=_FOUNDRY_API_BASE,
api_key="fake-foundry-key",
)
assert [str(call.request.url) for call in respx_mock.calls] == [f"{_FOUNDRY_API_BASE}/openai/v1/responses"]
request: Final = responses_route.calls[0].request
request_body: Final = json.loads(request.content)
assert request_body["tools"][0]["type"] == "function"
assert request_body["tools"][0]["name"] == "get_weather"
assert request.headers["api-key"] == "fake-foundry-key"
assert response.choices[0].finish_reason == "tool_calls"
assert response.choices[0].message.tool_calls[0].function.name == "get_weather"
@pytest.mark.parametrize(
"model, model_info, expected_model_param, expected_base_model_param",
[