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

Merge 9db0487383 into 767613bd40

Browse source
This commit is contained in:
tavren27 2026-11-04 05:07:16 +00:00 • committed by GitHub
commit 5b8fcf5939
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 1447 additions and 157 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
litellm/llms/anthropic/messages/adapters/handler.py
View file

@ -37,12 +37,14 @@ from litellm.llms.anthropic.messages.safeguards import (
with_safeguard_results,
)
from litellm.llms.anthropic.messages.utils import (
REASONING_EFFORT_OFF,
ProxySubCallLoggingKwargs,
is_reasoning_auto_summary_enabled,
litellm_logging_obj_from_kwargs,
local_model_name,
proxy_sub_call_logging_kwargs,
proxy_sub_call_trace_id,
requested_reasoning_summary,
)
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic_messages.anthropic_response import (
@ -468,29 +470,17 @@ class LiteLLMMessagesToCompletionTransformationHandler:
updated_reasoning_effort["summary"] = effective_summary
completion_kwargs["reasoning_effort"] = updated_reasoning_effort
@staticmethod
def _requested_reasoning_summary(thinking: Mapping[str, object]) -> str | None:
if thinking.get("type") == "disabled":
return None
summary: Final = thinking.get("summary")
if isinstance(summary, str):
return summary
display: Final = thinking.get("display")
return "auto" if display is not None and display != "omitted" else None
@staticmethod
def _chat_target_reasoning(
completion_kwargs: _CompletionKwargs,
*,
thinking: Mapping[str, object] | None,
) -> _ChatTargetReasoning | None:
if not thinking:
return None
reasoning_effort: Final = completion_kwargs.get("reasoning_effort")
if not thinking or reasoning_effort == REASONING_EFFORT_OFF:
return None
wraps_summary: Final = isinstance(reasoning_effort, dict) and "summary" in reasoning_effort
requested_summary: Final = LiteLLMMessagesToCompletionTransformationHandler._requested_reasoning_summary(
thinking
)
requested_summary: Final = requested_reasoning_summary(thinking)
if not wraps_summary and requested_summary is None:
return None
effort: Final[object] = (
@ -678,6 +668,8 @@ class LiteLLMMessagesToCompletionTransformationHandler:
custom_llm_provider=target_provider,
api_base=api_base if isinstance(api_base, str) else None,
drop_params=drop_params,
translate_bare_output_effort=extra_kwargs.get("reasoning_effort") is None,
deployment_model_info=extra_kwargs.get("model_info"),
)
if openai_request is None:

101
litellm/llms/anthropic/messages/adapters/transformation.py
View file

@ -12,9 +12,15 @@ from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.hidden_params import get_hidden_params
from litellm.llms.anthropic.messages.convertible_blocks import tool_result_text_part
from litellm.llms.anthropic.messages.utils import (
REASONING_EFFORT_OFF,
RESPONSES_API_PROVIDERS,
is_between_tool_calls,
is_reasoning_auto_summary_enabled,
json_copy_or_original,
maps_bare_output_effort,
prompt_cache_key_from_user_id,
reasoning_off_level,
thinking_off_switch,
)
from litellm.responses.utils import auto_reasoning_summary
@ -265,6 +271,8 @@ class AnthropicAdapter:
custom_llm_provider: str | None = None,
api_base: str | None = None,
drop_params: bool = True,
translate_bare_output_effort: bool = False,
deployment_model_info: object = None,
) -> tuple[ChatCompletionRequest | None, dict[str, str]]:
"""
Translate Anthropic request params to OpenAI format, returning tool name mapping.
@ -301,6 +309,8 @@ class AnthropicAdapter:
api_base=api_base,
drop_params=drop_params,
report_reasoning_drops=True,
translate_bare_output_effort=translate_bare_output_effort,
deployment_model_info=deployment_model_info,
)
return translated_body, tool_name_mapping
@ -790,7 +800,7 @@ class LiteLLMAnthropicMessagesAdapter:
reasoning_effort as a plain string, not a summary dict.
"""
thinking_type: Final = thinking.get("type") if isinstance(thinking, dict) else None
if thinking_type == "disabled":
if thinking_type == "disabled" or reasoning_effort == REASONING_EFFORT_OFF:
return reasoning_effort
summary: Final = thinking.get("summary") if isinstance(thinking, dict) else None
@ -1152,6 +1162,29 @@ class LiteLLMAnthropicMessagesAdapter:
new_kwargs["tools"] = translated_tools
return tool_name_mapping
@classmethod
def _is_claude_target(cls, model: str, custom_llm_provider: str | None) -> bool:
return (
cls.is_anthropic_claude_model(model) or cls.is_bedrock_arn_model(model)
) and custom_llm_provider not in _CLAUDE_OVER_REASONING_EFFORT_PROVIDERS
@staticmethod
def _bare_output_effort(
output_effort: object, model: str, custom_llm_provider: str | None, deployment_model_info: object
) -> str | None:
return (
output_effort
if isinstance(output_effort, str)
and maps_bare_output_effort(model, custom_llm_provider, deployment_model_info=deployment_model_info)
else None
)
@staticmethod
def _between_tools_reasoning_effort(anthropic_message_request: Mapping[str, object], output_effort: object) -> str:
if not is_between_tool_calls(anthropic_message_request):
return REASONING_EFFORT_OFF
return output_effort if isinstance(output_effort, str) and output_effort else "medium"
def _translate_thinking_to_openai(
self,
anthropic_message_request: AnthropicMessagesRequest,
@ -1160,6 +1193,8 @@ class LiteLLMAnthropicMessagesAdapter:
custom_llm_provider: str | None = None,
drop_params: bool = True,
report_reasoning_drops: bool = False,
translate_bare_output_effort: bool = False,
deployment_model_info: object = None,
) -> None:
"""Translate Anthropic thinking to either thinking or reasoning_effort.
@ -1185,23 +1220,16 @@ class LiteLLMAnthropicMessagesAdapter:
An adaptive request with no tier stays untouched either way, so the provider's own default
still applies.
"""
if "thinking" not in anthropic_message_request:
return
thinking: Final = anthropic_message_request["thinking"]
if not thinking:
return
model: Final = new_kwargs.get("model", "")
output_config: Final = anthropic_message_request.get("output_config")
output_effort: Final = output_config.get("effort") if isinstance(output_config, dict) else None
thinking: Final = anthropic_message_request.get("thinking")
is_bedrock_target: Final = model.startswith(("bedrock/", "converse/", "invoke/")) or self.is_bedrock_arn_model(
model
)
is_claude_target: Final = (
self.is_anthropic_claude_model(model) or self.is_bedrock_arn_model(model)
) and custom_llm_provider not in _CLAUDE_OVER_REASONING_EFFORT_PROVIDERS
output_config: Final = anthropic_message_request.get("output_config")
is_claude_target: Final = self._is_claude_target(model, custom_llm_provider)
if is_claude_target:
if thinking and is_claude_target:
new_kwargs["thinking"] = thinking
if is_bedrock_target:
if isinstance(output_config, dict):
@ -1213,28 +1241,51 @@ class LiteLLMAnthropicMessagesAdapter:
return
thinking_type: Final = thinking.get("type") if isinstance(thinking, dict) else None
declared_effort: Final = (
output_config.get("effort") if thinking_type == "adaptive" and isinstance(output_config, dict) else None
)
is_between_tools: Final = isinstance(thinking_type, str) and thinking_type == "between_tools"
declared_effort: Final = output_effort if thinking_type == "adaptive" else None
if is_claude_target and not declared_effort:
return
reasoning_effort: Final = declared_effort or self.translate_anthropic_thinking_to_reasoning_effort(
cast(AnthropicThinkingParam, thinking),
model=model,
drop_params=drop_params,
report_reasoning_drops=report_reasoning_drops,
mirrors_responses_bridge: Final = custom_llm_provider in RESPONSES_API_PROVIDERS
reasoning_effort: Final = declared_effort or (
self._between_tools_reasoning_effort(anthropic_message_request, output_effort)
if is_between_tools and mirrors_responses_bridge
else self.translate_anthropic_thinking_to_reasoning_effort(
cast(AnthropicThinkingParam, thinking),
model=model,
drop_params=drop_params,
report_reasoning_drops=report_reasoning_drops,
)
if thinking
else self._bare_output_effort(output_effort, model, custom_llm_provider, deployment_model_info)
if translate_bare_output_effort and mirrors_responses_bridge
else None
)
if not reasoning_effort:
return
new_kwargs["reasoning_effort"] = (
effort_sent: Final = (
reasoning_effort
if is_claude_target
if is_claude_target or not thinking
else self._apply_reasoning_summary_wrapping(
reasoning_effort, cast(dict[str, object], thinking), model, custom_llm_provider
)
)
off_switch: Final = thinking_off_switch(anthropic_message_request)
target_effort: Final = (
reasoning_off_level(
model,
custom_llm_provider,
deployment_model_info=deployment_model_info,
drop_params=drop_params,
between_tools=off_switch == "between_tools",
)
if mirrors_responses_bridge and off_switch is not None and effort_sent == REASONING_EFFORT_OFF
else effort_sent
)
if target_effort is None:
return
new_kwargs["reasoning_effort"] = target_effort
def _translate_output_format_to_openai(
self,
@ -1284,6 +1335,8 @@ class LiteLLMAnthropicMessagesAdapter:
preserve_midturn_system: bool = False,
drop_params: bool = True,
report_reasoning_drops: bool = False,
translate_bare_output_effort: bool = False,
deployment_model_info: object = None,
) -> tuple[ChatCompletionRequest, dict[str, str]]:
"""
This is used by the beta Anthropic Adapter, for translating anthropic `/v1/messages` requests to the openai format.
@ -1339,6 +1392,8 @@ class LiteLLMAnthropicMessagesAdapter:
custom_llm_provider=custom_llm_provider,
drop_params=drop_params,
report_reasoning_drops=report_reasoning_drops,
translate_bare_output_effort=translate_bare_output_effort,
deployment_model_info=deployment_model_info,
)
## CONVERT STOP_SEQUENCES
self._translate_stop_sequences_to_openai(

6
litellm/llms/anthropic/messages/native/handler.py
View file

@ -57,14 +57,12 @@ from ..safeguards import (
proxy_serves_safeguards,
with_safeguard_results,
)
from ..utils import RESPONSES_API_PROVIDERS
from .interceptors import get_messages_interceptors
from .utils import AnthropicMessagesRequestUtils, mock_response, prepare_native_messages
__all__ = ("anthropic_messages", "anthropic_messages_handler")
# Providers that are routed directly to the OpenAI Responses API instead of
# going through chat/completions.
_RESPONSES_API_PROVIDERS: Final = frozenset({"openai"})
_SAFEGUARD_KWARGS: Final = TypeAdapter(Mapping[str, object])
_SAFEGUARD_MESSAGES: Final = TypeAdapter(Sequence[Mapping[str, object]])
@ -174,7 +172,7 @@ def _should_route_to_responses_api(
return False
if stop_sequences:
return False
if custom_llm_provider in _RESPONSES_API_PROVIDERS:
if custom_llm_provider in RESPONSES_API_PROVIDERS:
return True
if custom_llm_provider is None or requested_model is None or resolved_model is None:
return False

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

@ -5,7 +5,7 @@ Used when the target model is an OpenAI or Azure model.
"""
from collections.abc import AsyncIterator, Callable, Coroutine, Mapping, Sequence
from typing import Final, TypeAlias
from typing import Final, TypeAlias, get_args
from pydantic import TypeAdapter, ValidationError
@ -25,7 +25,9 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
prompt_cache_breakpoint_with_readable_mode,
with_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.reasoning_effort_capability import deployment_rounds_reasoning_effort
from litellm.litellm_core_utils.tool_fields import drops_mcp_servers, tools_without_mcp_servers
from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfig
from litellm.responses.utils import auto_reasoning_summary
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic import (
@ -39,8 +41,8 @@ from litellm.types.llms.anthropic import (
from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
)
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import CountContextManagement
from litellm.types.llms.openai import REASONING_EFFORT, ResponsesAPIResponse
from litellm.types.utils import CountContextManagement, LlmProviders
from litellm.utils import ProviderConfigManager
from ..adapters.handler import context_management_explicitly_dropped, request_safeguards_evaluator
@ -49,7 +51,17 @@ from ..context_management.constants import COMPACT_EDIT_TYPE
from ..context_management.dispatcher import normalize_spec
from ..convertible_blocks import reject_malformed_tool_result_text
from ..safeguards import with_safeguard_results
from ..utils import litellm_logging_obj_from_kwargs, local_model_name
from ..utils import (
REASONING_EFFORT_OFF,
ThinkingOffSwitch,
deployment_reasoning_efforts,
litellm_logging_obj_from_kwargs,
local_model_name,
maps_bare_output_effort,
normalize_reasoning_effort_value,
reasoning_off_level,
thinking_off_switch,
)
from .streaming_iterator import AnthropicResponsesStreamWrapper
from .transformation import LiteLLMAnthropicToResponsesAPIAdapter
@ -62,6 +74,8 @@ _SplitEdits: TypeAlias = tuple[dict[str, object] | None, Sequence[Mapping[str, o
AnthropicContextManagementSpec: TypeAlias = dict[str, object] | list[dict[str, object]] | None
_ADAPTER: Final = LiteLLMAnthropicToResponsesAPIAdapter()
_KNOWN_REASONING_EFFORTS: Final = frozenset(get_args(REASONING_EFFORT))
_UNVERIFIED_ORGANIZATION_SUMMARY_REFUSAL: Final = "must be verified to generate reasoning summaries"
def _split_context_management(
@ -251,6 +265,119 @@ def _rewritten_blocks(content: object, rewrite: BlockRewrite) -> Sequence[Mappin
return ()
def _reasoning_for_target(
reasoning: object,
*,
model: str,
custom_llm_provider: str | None,
deployment_model_info: object,
drop_params: bool,
off_switch: ThinkingOffSwitch | None,
) -> object:
reasoning_params: Final = as_str_mapping(reasoning)
effort: Final = reasoning_params.get("effort") if reasoning_params is not None else None
if reasoning_params is None or not isinstance(effort, str):
return reasoning
if effort == REASONING_EFFORT_OFF and off_switch is not None:
off_level: Final = reasoning_off_level(
model,
custom_llm_provider,
deployment_model_info=deployment_model_info,
drop_params=drop_params,
between_tools=off_switch == "between_tools",
)
if off_level is None or _responses_api_refuses_reasoning_effort(model, custom_llm_provider):
return None
return {**reasoning_params, "effort": off_level}
if drop_params and effort not in _KNOWN_REASONING_EFFORTS:
verbose_logger.warning(
"Dropping reasoning.effort=%r for %s, no reasoning API defines it, because drop_params is set",
effort,
model,
)
return None
normalized: Final = normalize_reasoning_effort_value(
effort,
model=model,
custom_llm_provider=custom_llm_provider,
deployment_model_info=deployment_model_info,
)
if drop_params and _model_excludes_effort(
normalized, model=model, custom_llm_provider=custom_llm_provider, deployment_model_info=deployment_model_info
):
verbose_logger.warning(
"Dropping reasoning.effort=%r for %s, its model cost map entry does not list it, because drop_params is set",
normalized,
model,
)
return None
return reasoning if normalized == effort else {**reasoning_params, "effort": normalized}
def _responses_api_refuses_reasoning_effort(model: str, custom_llm_provider: str | None) -> bool:
return custom_llm_provider == LlmProviders.OPENAI.value and not OpenAIResponsesAPIConfig.supports_reasoning_param(
model
)
def _model_excludes_effort(
effort: str, *, model: str, custom_llm_provider: str | None, deployment_model_info: object
) -> bool:
if deployment_rounds_reasoning_effort(deployment_model_info):
return False
levels: Final = deployment_reasoning_efforts(model, custom_llm_provider, deployment_model_info) or ()
return bool(levels) and effort not in levels
def _retry_without_refused_summary(
responses_kwargs: Mapping[str, object], error: litellm.BadRequestError, *, model: str, drop_params: bool
) -> Mapping[str, object] | None:
reasoning: Final = as_str_mapping(responses_kwargs.get("reasoning"))
if not drop_params or reasoning is None or "summary" not in reasoning:
return None
if _UNVERIFIED_ORGANIZATION_SUMMARY_REFUSAL not in error.message:
return None
verbose_logger.warning(
"Retrying %s without reasoning.summary, its OpenAI organization cannot get reasoning summaries until it is "
"verified, because drop_params is set",
model,
)
return {**responses_kwargs, "reasoning": {key: value for key, value in reasoning.items() if key != "summary"}}
async def _aresponses(responses_kwargs: Mapping[str, object], *, model: str, drop_params: bool) -> object:
try:
return await litellm.aresponses(**responses_kwargs)
except litellm.BadRequestError as error:
retry_kwargs: Final = _retry_without_refused_summary(
responses_kwargs, error, model=model, drop_params=drop_params
)
if retry_kwargs is None:
raise
return await litellm.aresponses(**retry_kwargs)
def _responses(responses_kwargs: Mapping[str, object], *, model: str, drop_params: bool) -> object:
try:
return litellm.responses(**responses_kwargs)
except litellm.BadRequestError as error:
retry_kwargs: Final = _retry_without_refused_summary(
responses_kwargs, error, model=model, drop_params=drop_params
)
if retry_kwargs is None:
raise
return litellm.responses(**retry_kwargs)
def _maps_reasoning(
thinking: Mapping[str, object] | None, model: str, provider: str | None, forwarded_kwargs: Mapping[str, object]
) -> bool:
return bool(thinking) or (
forwarded_kwargs.get("reasoning_effort") is None
and maps_bare_output_effort(model, provider, deployment_model_info=forwarded_kwargs.get("model_info"))
)
def _build_responses_kwargs(
*,
max_tokens: int,
@ -326,34 +453,33 @@ def _build_responses_kwargs(
request_data["output_format"] = output_format
anthropic_request: Final = AnthropicMessagesRequest(**request_data)
drop_params: Final = drops_unsupported_params(forwarded_kwargs.get("drop_params"))
responses_kwargs: Final = _ADAPTER.translate_request(
anthropic_request,
include_encrypted_reasoning=_provider_returns_encrypted_reasoning(
model, forwarded_kwargs.get("custom_llm_provider")
),
auto_summary_level=auto_reasoning_summary(model, forwarded_kwargs.get("custom_llm_provider")),
drop_params=drops_unsupported_params(forwarded_kwargs.get("drop_params")),
drop_params=drop_params,
)
# Normalize reasoning effort based on model capabilities
# (e.g. "max" → "xhigh"/"high", "minimal" → "low" if unsupported)
reasoning: Final = responses_kwargs.get("reasoning")
if isinstance(reasoning, dict):
effort: Final[object] = reasoning.get("effort")
if isinstance(effort, str):
from litellm.llms.anthropic.messages.utils import (
normalize_reasoning_effort_value,
)
provider_hint: Final = forwarded_kwargs.get("custom_llm_provider")
normalized: Final = normalize_reasoning_effort_value(
effort,
model=model,
custom_llm_provider=provider_hint if isinstance(provider_hint, str) else None,
deployment_model_info=forwarded_kwargs.get("model_info"),
)
if normalized != effort:
responses_kwargs["reasoning"] = {**reasoning, "effort": normalized}
provider_hint: Final = forwarded_kwargs.get("custom_llm_provider")
provider: Final = provider_hint if isinstance(provider_hint, str) else None
bridged_reasoning: Final = responses_kwargs.pop("reasoning", None)
target_reasoning: Final = (
_reasoning_for_target(
bridged_reasoning,
model=model,
custom_llm_provider=provider,
deployment_model_info=forwarded_kwargs.get("model_info"),
drop_params=drop_params,
off_switch=thinking_off_switch(anthropic_request),
)
if _maps_reasoning(thinking, model, provider, forwarded_kwargs)
else None
)
if target_reasoning is not None:
responses_kwargs["reasoning"] = target_reasoning
if stream:
responses_kwargs["stream"] = True
@ -443,7 +569,9 @@ class LiteLLMMessagesToResponsesAPIHandler:
)
safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
result: Final = await litellm.aresponses(**responses_kwargs)
result: Final = await _aresponses(
responses_kwargs, model=model, drop_params=drops_unsupported_params(kwargs.get("drop_params"))
)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(
@ -546,7 +674,9 @@ class LiteLLMMessagesToResponsesAPIHandler:
extra_kwargs=kwargs,
)
result: Final = litellm.responses(**responses_kwargs)
result: Final = _responses(
responses_kwargs, model=model, drop_params=drops_unsupported_params(kwargs.get("drop_params"))
)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(

125
litellm/llms/anthropic/messages/responses_adapters/transformation.py
View file

@ -10,6 +10,7 @@ from collections.abc import Iterable, Mapping, Sequence
from itertools import groupby
from typing import Any, Final, cast
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_BOUNDARY,
TOOL_RESULT_IMAGE_PLACEHOLDER,
@ -27,9 +28,13 @@ from litellm.llms.anthropic.messages.native.utils import (
responses_output_refusal_text,
)
from litellm.llms.anthropic.messages.utils import (
REASONING_EFFORT_OFF,
content_blocks,
is_between_tool_calls,
is_reasoning_auto_summary_enabled,
json_copy_or_original,
prompt_cache_key_from_user_id,
requested_reasoning_summary,
)
from litellm.types.llms.anthropic import (
AllAnthropicPassThroughMessageValues,
@ -125,16 +130,6 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
return {"type": "input_file", "file_url": url}
return None
@staticmethod
def _content_blocks(content: object) -> tuple[Mapping[str, object], ...]:
if not isinstance(content, list):
return ()
return tuple(
cast(Mapping[str, object], block) # cast-ok: untrusted client payload
for block in cast(Sequence[object], content) # cast-ok: untyped client content list
if isinstance(block, dict)
)
@classmethod
def _tool_result_image_parts(
cls, blocks: Sequence[Mapping[str, object]]
@ -159,7 +154,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
block, the images Anthropic caches for the mark. The result's text and documents join into one output, which
the mark covers whole.
"""
inner: Final = cls._content_blocks(block.get("content"))
inner: Final = content_blocks(block.get("content"))
own_breakpoint: Final = block.get("prompt_cache_breakpoint")
if own_breakpoint is not None:
return own_breakpoint, cls._tool_result_image_parts(inner), ()
@ -529,54 +524,73 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
return result if result else None
@staticmethod
def _requested_effort(
thinking: Mapping[str, object],
configured_effort: object,
*,
model: str,
drop_params: bool,
between_tool_calls: bool,
) -> object:
match thinking.get("type"):
case "adaptive":
return configured_effort or "medium"
case "enabled":
return reasoning_effort_from_thinking(thinking, model=model, drop_params=drop_params)
case "disabled":
return REASONING_EFFORT_OFF
case "between_tools":
return (configured_effort or "medium") if between_tool_calls else REASONING_EFFORT_OFF
case _:
return configured_effort or None
@staticmethod
def _reasoning_summary(thinking: Mapping[str, object], auto_summary_level: str) -> str | None:
requested: Final = requested_reasoning_summary(thinking)
if requested:
return requested
if thinking.get("display") is not None:
return None
return auto_summary_level if is_reasoning_auto_summary_enabled() else None
@classmethod
def translate_thinking_to_reasoning(
thinking: dict[str, object],
output_config: dict[str, object] | None = None,
cls,
thinking: object,
output_config: object = None,
auto_summary_level: str = "detailed",
*,
model: str,
drop_params: bool = True,
between_tool_calls: bool = False,
) -> dict[str, object] | None:
"""
Convert Anthropic thinking param to Responses API reasoning param.
Convert Anthropic thinking and ``output_config.effort`` to the Responses API reasoning param.
``thinking.budget_tokens`` is bucketed via the shared
``reasoning_effort_from_thinking_budget`` thresholds. For adaptive
thinking, uses ``output_config.effort`` if available, otherwise defaults
to medium.
``between_tools`` thinking reasons only once the model has tool results to read, so a fresh turn turns
reasoning off like ``disabled`` does.
"""
if not isinstance(thinking, dict):
thinking_params: Final[Mapping[str, object]] = as_str_mapping(thinking) or {}
output_params: Final = as_str_mapping(output_config)
effort: Final = cls._requested_effort(
thinking_params,
output_params.get("effort") if output_params is not None else None,
model=model,
drop_params=drop_params,
between_tool_calls=between_tool_calls,
)
if effort is None:
return None
thinking_type: Final = thinking.get("type")
if thinking_type == "adaptive":
# Use output_config.effort if available
effort = "medium"
if isinstance(output_config, dict) and output_config.get("effort"):
effort = output_config["effort"]
elif thinking_type == "enabled":
budget_effort: Final = reasoning_effort_from_thinking(thinking, model=model, drop_params=drop_params)
if budget_effort is None:
return None
effort = budget_effort
else:
return None
auto_summary: Final = is_reasoning_auto_summary_enabled()
result: Final[dict[str, object]] = {"effort": effort}
summary: Final = thinking.get("summary")
if summary:
result["summary"] = summary
elif auto_summary:
result["summary"] = auto_summary_level
return result
if effort == REASONING_EFFORT_OFF:
return {"effort": effort}
summary: Final = cls._reasoning_summary(thinking_params, auto_summary_level)
return {"effort": effort} if summary is None else {"effort": effort, "summary": summary}
@classmethod
def system_blocks_to_instructions(cls, system: object) -> str:
return "\n".join(
text
for block in cls._content_blocks(system)
for block in content_blocks(system)
if block.get("type") == "text" and isinstance(text := block.get("text"), str) and text
)
@ -660,19 +674,16 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
cast(AnthropicMessagesToolChoice, tool_choice)
)
# thinking -> reasoning
thinking: Final = anthropic_request.get("thinking")
if isinstance(thinking, dict):
output_config = anthropic_request.get("output_config")
reasoning: Final = self.translate_thinking_to_reasoning(
thinking,
output_config=cast(dict[str, object] | None, output_config),
auto_summary_level=auto_summary_level,
model=model,
drop_params=drop_params,
)
if reasoning:
responses_kwargs["reasoning"] = reasoning
reasoning: Final = self.translate_thinking_to_reasoning(
anthropic_request.get("thinking"),
output_config=anthropic_request.get("output_config"),
auto_summary_level=auto_summary_level,
model=model,
drop_params=drop_params,
between_tool_calls=is_between_tool_calls(anthropic_request),
)
if reasoning:
responses_kwargs["reasoning"] = reasoning
# output_format / output_config.format -> text format
# output_format: {"type": "json_schema", "schema": {...}}

182
litellm/llms/anthropic/messages/utils.py
View file

@ -1,14 +1,17 @@
import json
import os
from collections.abc import Mapping
from collections.abc import Iterator, Mapping, Sequence
from itertools import chain
from types import MappingProxyType
from typing import TYPE_CHECKING, Final, TypeVar
from typing import TYPE_CHECKING, Final, Literal, TypeVar
from pydantic import ConfigDict, TypeAdapter, ValidationError
from typing_extensions import ReadOnly
import litellm
from litellm._logging import verbose_logger
from litellm.constants import X_LITELLM_DISABLE_CALLBACKS
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.internal_call_metadata import sanitize_user_api_key_auth
from litellm.types.llms.base import LiteLLMBaseModel
from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD, ModelInfo, StandardCallbackDynamicParams
@ -25,7 +28,12 @@ _EFFORT_DEGRADATION_CHAIN: Final[Mapping[str, tuple[str, ...]]] = MappingProxyTy
"minimal": ("minimal", "low"),
}
)
_THINKING_OFF: Final = "none"
REASONING_EFFORT_OFF: Final = "none"
RESPONSES_API_PROVIDERS: Final = frozenset({"openai"})
_LOWEST_REASONING_TIERS: Final = ("minimal", "low")
_UNCONFIRMED_MINIMAL_LOWEST_TIERS: Final = ("low",)
ThinkingOffSwitch = Literal["disabled", "between_tools"]
_JsonValue: Final = TypeVar("_JsonValue")
@ -83,6 +91,7 @@ PROXY_SUB_CALL_METADATA_KEYS: Final = (
)
_USER_API_KEY_AUTH_METADATA_KEY: Final = "user_api_key_auth"
_OBJECT_MAPPING: Final = TypeAdapter(Mapping[str, object])
_OBJECT_LIST: Final = TypeAdapter(list[object])
_NO_SETTINGS: Final[Mapping[str, object]] = MappingProxyType({})
@ -189,6 +198,158 @@ def is_reasoning_auto_summary_enabled() -> bool:
return litellm.reasoning_auto_summary or os.getenv("LITELLM_REASONING_AUTO_SUMMARY", "false").lower() == "true"
def requested_reasoning_summary(thinking: Mapping[str, object]) -> str | None:
if thinking.get("type") == "disabled":
return None
summary: Final = thinking.get("summary")
if isinstance(summary, str):
return summary
display: Final = thinking.get("display")
if display is not None:
return None if display == "omitted" else "auto"
return "auto" if thinking.get("type") == "between_tools" else None
def _list_items(value: object) -> Sequence[object]:
try:
return _OBJECT_LIST.validate_python(value, strict=True)
except ValidationError:
return ()
def content_blocks(content: object) -> tuple[Mapping[str, object], ...]:
return tuple(block for block in map(as_str_mapping, _list_items(content)) if block is not None)
def _final_user_turn(messages: object) -> Iterator[Mapping[str, object]]:
for message in map(as_str_mapping, reversed(_list_items(messages))):
if message is None or message.get("role") != "user":
return
yield message
def is_between_tool_calls(request: Mapping[str, object]) -> bool:
blocks: Final = chain.from_iterable(
content_blocks(message.get("content")) for message in _final_user_turn(request.get("messages"))
)
return any(block.get("type") == "tool_result" for block in blocks)
def thinking_off_switch(request: Mapping[str, object]) -> ThinkingOffSwitch | None:
thinking: Final = as_str_mapping(request.get("thinking"))
thinking_type: Final = thinking.get("type") if thinking is not None else None
if thinking_type == "disabled":
return "disabled"
if thinking_type == "between_tools" and not is_between_tool_calls(request):
return "between_tools"
return None
def _mapped_model_info(model: str, custom_llm_provider: str | None) -> ModelInfo | None:
from litellm.utils import get_model_info
try:
return get_model_info(model=model, custom_llm_provider=custom_llm_provider)
except Exception:
return None
def _deployment_model_info(
model: str, custom_llm_provider: str | None, deployment_model_info: object
) -> Mapping[str, object] | None:
from litellm.utils import catalog_entry
own_info: Final = as_str_mapping(deployment_model_info)
if own_info is None:
return _mapped_model_info(model, custom_llm_provider)
declared: Final = {key: value for key, value in own_info.items() if value is not None}
return {**(catalog_entry(model, custom_llm_provider) or {}), **declared}
def _resolved_reasoning_efforts(
model_info: Mapping[str, object] | None, model: str, custom_llm_provider: str | None
) -> tuple[str, ...] | None:
from litellm.litellm_core_utils.reasoning_effort_capability import resolve_supported_reasoning_efforts
from litellm.utils import has_catalog_entry
if model_info is None:
return None
return resolve_supported_reasoning_efforts(
model_info, deployment_is_mapped=has_catalog_entry(model, custom_llm_provider)
)
def model_reasoning_efforts(model: str, custom_llm_provider: str | None) -> tuple[str, ...] | None:
return _resolved_reasoning_efforts(_mapped_model_info(model, custom_llm_provider), model, custom_llm_provider)
def deployment_reasoning_efforts(
model: str, custom_llm_provider: str | None, deployment_model_info: object
) -> tuple[str, ...] | None:
return _resolved_reasoning_efforts(
_deployment_model_info(model, custom_llm_provider, deployment_model_info), model, custom_llm_provider
)
def _declares(model_info: Mapping[str, object] | None, capability: str) -> bool:
return model_info is not None and model_info.get(capability) is True
def _confirms_minimal(model_info: Mapping[str, object] | None) -> bool:
from litellm.litellm_core_utils.reasoning_effort_capability import declared_reasoning_efforts
return model_info is not None and (
declared_reasoning_efforts(model_info) is not None or _declares(model_info, "supports_minimal_reasoning_effort")
)
def _lowest_tier_instead_of_off(levels: tuple[str, ...] | None, model_info: Mapping[str, object] | None) -> str | None:
if not levels or REASONING_EFFORT_OFF in levels:
return None
tiers: Final = _LOWEST_REASONING_TIERS if _confirms_minimal(model_info) else _UNCONFIRMED_MINIMAL_LOWEST_TIERS
return next((tier for tier in tiers if tier in levels), None)
def reasoning_off_level(
model: str,
custom_llm_provider: str | None,
*,
deployment_model_info: object,
drop_params: bool,
between_tools: bool,
) -> str | None:
model_info: Final = _deployment_model_info(model, custom_llm_provider, deployment_model_info)
levels: Final = _resolved_reasoning_efforts(model_info, model, custom_llm_provider)
if levels == () or (between_tools and levels is None):
return None
lowest_tier: Final = _lowest_tier_instead_of_off(levels, model_info)
if lowest_tier is not None:
return lowest_tier
if (
not drop_params
or not _declares(model_info, "supports_reasoning")
or _declares(model_info, "supports_none_reasoning_effort")
):
return REASONING_EFFORT_OFF
verbose_logger.warning(
"Dropping reasoning_effort=none for %s, its model cost map entry does not set "
"supports_none_reasoning_effort, because drop_params is set",
model,
)
return None
def maps_bare_output_effort(model: str, custom_llm_provider: str | None, *, deployment_model_info: object) -> bool:
from litellm.litellm_core_utils.reasoning_effort_capability import resolve_supported_reasoning_efforts
model_info: Final = _deployment_model_info(model, custom_llm_provider, deployment_model_info)
return (
model_info is not None
and model_info.get("default_reasoning_effort") != REASONING_EFFORT_OFF
and resolve_supported_reasoning_efforts(model_info, deployment_is_mapped=True) != ()
)
def normalize_reasoning_effort_value(
effort: str,
model: str,
@ -214,23 +375,14 @@ def normalize_reasoning_effort_value(
if chain is None:
return effort
from litellm.litellm_core_utils.reasoning_effort_capability import (
deployment_rounds_reasoning_effort,
resolve_supported_reasoning_efforts,
)
from litellm.litellm_core_utils.reasoning_effort_capability import deployment_rounds_reasoning_effort
if deployment_rounds_reasoning_effort(deployment_model_info):
return effort
from litellm.utils import get_model_info
try:
model_info: Final[ModelInfo] = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
except Exception:
return chain[-1]
supported: Final = resolve_supported_reasoning_efforts(model_info, deployment_is_mapped=True)
supported: Final = model_reasoning_efforts(model, custom_llm_provider)
if not supported:
return chain[-1]
accepted_tiers: Final = tuple(level for level in supported if level != _THINKING_OFF)
accepted_tiers: Final = tuple(level for level in supported if level != REASONING_EFFORT_OFF)
return next((level for level in (*chain, *accepted_tiers) if level in supported), chain[-1])

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

@ -405,7 +405,7 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
return OpenAIGPT5Config.effort_resolves_to_none(model, effort)
@staticmethod
def _supports_reasoning_param(model: str) -> bool:
def supports_reasoning_param(model: str) -> bool:
from litellm.utils import get_model_info_helper
try:
@ -503,7 +503,7 @@ class OpenAIResponsesAPIConfig(BaseResponsesAPIConfig):
if (
self.custom_llm_provider == LlmProviders.OPENAI
and self._requests_reasoning_effort(params.get("reasoning"))
and not self._supports_reasoning_param(model=model)
and not self.supports_reasoning_param(model=model)
):
if drop_params or litellm.drop_params:
dropped_reasoning: Final = params.pop("reasoning", None)

8
litellm/utils.py
View file

@ -5478,9 +5478,13 @@ def is_known_non_reasoning_model(model: str, custom_llm_provider: str | None) ->
return info.get("supports_reasoning") is False or info["key"] in GetModelCostMap.loaded_model_cost_map()
def has_catalog_entry(model: str, custom_llm_provider: str | None) -> bool:
def catalog_entry(model: str, custom_llm_provider: str | None) -> Mapping[str, object] | None:
info: Final = lookup_catalog_model_info(model, custom_llm_provider)
return not isinstance(info, ModelInfoLookupMiss) and info["key"] in GetModelCostMap.loaded_model_cost_map()
return None if isinstance(info, ModelInfoLookupMiss) else GetModelCostMap.loaded_model_cost_map().get(info["key"])
def has_catalog_entry(model: str, custom_llm_provider: str | None) -> bool:
return catalog_entry(model, custom_llm_provider) is not None
def _dropped_reasoning_warning(model: str, custom_llm_provider: str | None) -> str:

2
tests/integration/translation/templates/openai.py
View file

@ -140,7 +140,7 @@ OPENAI_RESPONSES_REQUEST_FROM_MESSAGES: Final[Mapping[str, JsonValue]] = {
"zero, negative and non-integer quantities before choosing the next file to "
"inspect.",
"max_output_tokens": 1024,
"reasoning": {"effort": "high"},
"reasoning": {"effort": "high", "summary": "auto"},
"tools": [
{
"type": "function",

458
tests/unit/llms/anthropic/messages/adapters/test_handler_reasoning_effort_normalization.py
View file

@ -20,9 +20,10 @@ import litellm
from litellm.llms.anthropic.messages.adapters.handler import (
LiteLLMMessagesToCompletionTransformationHandler,
)
from litellm.llms.anthropic.messages.utils import maps_bare_output_effort, model_reasoning_efforts
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexGeminiConfig
from litellm.router import Router
from litellm.utils import reasoning_effort_with_summary_alias
from litellm.utils import reasoning_effort_with_summary_alias, supports_none_reasoning_effort, supports_reasoning
MESSAGES = [{"role": "user", "content": "hello"}]
@ -387,7 +388,7 @@ _RESPONSE: Final = {
"id": "msg_1",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "ok", "annotations": []}],
"content": [{"type": "output_text", "text": "ok", "annotations": ()}],
}
],
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
@ -582,3 +583,456 @@ class TestADeclaredDeploymentRoundsTheMessagesTierOntoItsLevels:
sent: Final = json.loads(route.calls.last.request.read())
assert functools.reduce(lambda body, key: body[key], effort_path, sent) == "minimal"
TOOL_RESULT_TURN: Final = [
{"role": "user", "content": "multiply 6 by 7"},
{
"role": "assistant",
"content": [{"type": "tool_use", "id": "toolu_1", "name": "multiply", "input": {"a": 6, "b": 7}}],
},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "42"}]},
]
OPENAI_CHAT_TARGETS: Final = ["openai/gpt-5.1", "openai/gpt-5-mini"]
OTHER_PROVIDER_CHAT_TARGETS: Final = [
("databricks/databricks-qwen35-122b-a10b", "databricks"),
("fireworks_ai/kimi-k3", "fireworks_ai"),
("gemini/gemini-3.1-pro-preview", "gemini"),
("hosted_vllm/unmapped-model", "hosted_vllm"),
]
def _chat_completion_kwargs(
model: str,
provider: str,
*,
thinking: dict[str, object] | None,
messages: list[dict[str, object]],
output_config: dict[str, str] | None = None,
reasoning_effort: str | None = None,
drop_params: bool | None = None,
) -> Mapping[str, object]:
extra_kwargs: Final = {
key: value
for key, value in (
("custom_llm_provider", provider),
("output_config", output_config),
("reasoning_effort", reasoning_effort),
("drop_params", drop_params),
)
if value is not None
}
completion_kwargs, _ = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
max_tokens=1024,
messages=messages,
model=model,
thinking=thinking,
tools=[MULTIPLY_TOOL],
extra_kwargs=extra_kwargs,
)
return completion_kwargs
def _lowest_accepted_level(model: str, *, drop_params: bool, between_tools: bool) -> str | None:
levels: Final = model_reasoning_efforts(model, "openai")
if levels is not None:
return levels[0] if levels else None
if between_tools:
return None
drops_unconfirmed_off: Final = (
drop_params and supports_reasoning(model, "openai") and not supports_none_reasoning_effort(model, "openai")
)
return None if drops_unconfirmed_off else "none"
class TestAnOpenAIChatRouteRequestGetsTheResponsesBridgeThinkingControls:
@pytest.mark.parametrize("model", OPENAI_CHAT_TARGETS)
@pytest.mark.parametrize("output_config, tier", [(None, "medium"), ({"effort": "high"}, "high")])
def test_between_tools_reasons_after_a_tool_result_at_the_declared_tier_or_medium(
self, local_model_cost_map: None, model: str, output_config: dict[str, str] | None, tier: str
) -> None:
sent: Final = _chat_completion_kwargs(
model, "openai", thinking={"type": "between_tools"}, messages=TOOL_RESULT_TURN, output_config=output_config
)
assert sent["reasoning_effort"] == tier
@pytest.mark.parametrize(
"model", [*OPENAI_CHAT_TARGETS, "openai/gpt-6.1-sol", "openai/gpt-4.1", "openai/o3"]
)
@pytest.mark.parametrize(
"thinking",
[{"type": "disabled"}, {"type": "between_tools", "summary": "concise"}],
ids=["disabled", "between_tools-first-turn"],
)
@pytest.mark.parametrize("drop_params", [False, True], ids=["no-drop_params", "drop_params"])
def test_reasoning_off_is_the_lowest_level_the_model_accepts(
self, local_model_cost_map: None, model: str, thinking: dict[str, object], drop_params: bool
) -> None:
sent: Final = _chat_completion_kwargs(
model,
"openai",
thinking=thinking,
messages=MESSAGES,
output_config={"effort": "high"},
drop_params=drop_params,
)
assert sent.get("reasoning_effort") == _lowest_accepted_level(
model, drop_params=drop_params, between_tools=thinking["type"] == "between_tools"
)
@pytest.mark.parametrize("model", [*OPENAI_CHAT_TARGETS, "openai/gpt-4.1", "openai/o3"])
def test_a_bare_output_effort_is_the_tier_when_the_model_reasons_unasked(
self, local_model_cost_map: None, model: str
) -> None:
sent: Final = _chat_completion_kwargs(
model, "openai", thinking=None, messages=MESSAGES, output_config={"effort": "high"}
)
assert sent.get("reasoning_effort") == (
"high" if maps_bare_output_effort(model, "openai", deployment_model_info=None) else None
)
def test_an_explicit_reasoning_effort_beats_a_bare_output_effort(self, local_model_cost_map: None) -> None:
sent: Final = _chat_completion_kwargs(
"openai/gpt-5-mini",
"openai",
thinking=None,
messages=MESSAGES,
output_config={"effort": "high"},
reasoning_effort="low",
)
assert sent["reasoning_effort"] == "low"
@pytest.mark.parametrize("model, provider", OTHER_PROVIDER_CHAT_TARGETS)
@pytest.mark.parametrize(
"thinking, messages",
[(None, MESSAGES), ({"type": "between_tools"}, TOOL_RESULT_TURN), ({"type": "between_tools"}, MESSAGES)],
ids=["bare-effort", "between_tools-after-a-tool-result", "between_tools-first-turn"],
)
def test_another_provider_keeps_ignoring_between_tools_and_a_bare_effort(
self,
local_model_cost_map: None,
model: str,
provider: str,
thinking: dict[str, object] | None,
messages: list[dict[str, object]],
) -> None:
sent: Final = _chat_completion_kwargs(
model, provider, thinking=thinking, messages=messages, output_config={"effort": "high"}
)
assert "reasoning_effort" not in sent
@pytest.mark.parametrize(
"thinking, messages",
[(None, MESSAGES), ({"type": "between_tools"}, TOOL_RESULT_TURN), ({"type": "between_tools"}, MESSAGES)],
ids=["bare-effort", "between_tools-after-a-tool-result", "between_tools-first-turn"],
)
def test_a_claude_target_still_gets_no_reasoning_effort(
self, local_model_cost_map: None, thinking: dict[str, object] | None, messages: list[dict[str, object]]
) -> None:
sent: Final = _chat_completion_kwargs(
"anthropic/claude-sonnet-4-5",
"anthropic",
thinking=thinking,
messages=messages,
output_config={"effort": "high"},
)
assert "reasoning_effort" not in sent
_OFF_MAP_OPENAI_DEPLOYMENT: Final = {
"model": "openai/off-map-hybrid-reasoner",
"api_base": "https://compatible.example.test/v1",
"api_key": "fake-key",
}
class TestDisabledOnAnOffMapOpenAIDeployment:
@pytest.fixture(autouse=True)
def _httpx_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
@pytest.mark.asyncio
async def test_the_chat_route_sends_reasoning_effort_none(
self, respx_mock: respx.MockRouter, local_model_cost_map: None
) -> None:
sent: Final = await _body_sent_upstream(
respx_mock,
_OFF_MAP_OPENAI_DEPLOYMENT,
{},
r"https://compatible\.example\.test/v1/chat/completions",
_CHAT_COMPLETION,
thinking={"type": "disabled"},
stop_sequences=["STOP"],
)
assert sent["reasoning_effort"] == "none"
@pytest.mark.asyncio
async def test_the_responses_bridge_sends_no_reasoning_its_transformation_would_refuse(
self, respx_mock: respx.MockRouter, local_model_cost_map: None
) -> None:
sent: Final = await _body_sent_upstream(
respx_mock,
_OFF_MAP_OPENAI_DEPLOYMENT,
{},
r"https://compatible\.example\.test/v1/responses",
_RESPONSE,
thinking={"type": "disabled"},
)
assert "reasoning" not in sent
@pytest.mark.asyncio
async def test_the_chat_route_sends_none_when_the_operator_sets_the_deployment_mode(
self, respx_mock: respx.MockRouter, local_model_cost_map: None
) -> None:
sent: Final = await _body_sent_upstream(
respx_mock,
{**_OFF_MAP_OPENAI_DEPLOYMENT, "model": "openai/vendor/off-map-hybrid-reasoner"},
{"mode": "responses"},
r"https://compatible\.example\.test/v1/responses",
_RESPONSE,
thinking={"type": "disabled"},
stop_sequences=["STOP"],
)
assert sent["reasoning"]["effort"] == "none"
@pytest.mark.asyncio
@pytest.mark.parametrize(
"messages_params",
[{"thinking": {"type": "disabled"}}, {"thinking": {"type": "adaptive"}, "output_config": {"effort": "none"}}],
ids=["disabled", "adaptive-effort-none"],
)
async def test_the_responses_bridge_sends_none_where_its_transformation_accepts_it(
self, respx_mock: respx.MockRouter, local_model_cost_map: None, messages_params: Mapping[str, object]
) -> None:
sent: Final = await _body_sent_upstream(
respx_mock,
{**_OFF_MAP_OPENAI_DEPLOYMENT, "model": "openai/vendor/off-map-hybrid-reasoner"},
{},
r"https://compatible\.example\.test/v1/responses",
_RESPONSE,
**messages_params,
)
assert sent["reasoning"] == {"effort": "none"}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"uses_chat_route, reasoning_key", [(True, "reasoning_effort"), (False, "reasoning")], ids=["chat-route", "responses-bridge"]
)
async def test_a_first_turn_between_tools_leaves_reasoning_to_the_model(
self, respx_mock: respx.MockRouter, local_model_cost_map: None, uses_chat_route: bool, reasoning_key: str
) -> None:
sent: Final = await _body_sent_upstream(
respx_mock,
{**_OFF_MAP_OPENAI_DEPLOYMENT, "model": "openai/vendor/off-map-hybrid-reasoner"},
{},
rf"https://compatible\.example\.test/v1/{'chat/completions' if uses_chat_route else 'responses'}",
_CHAT_COMPLETION if uses_chat_route else _RESPONSE,
thinking={"type": "between_tools"},
stop_sequences=["STOP"] if uses_chat_route else None,
)
assert reasoning_key not in sent
async def _bodies_sent_alone_and_beside_a_sibling(
respx_mock: respx.MockRouter,
litellm_params: Mapping[str, object],
sibling_model_info: Mapping[str, object],
upstream_url: str,
upstream_body: Mapping[str, object],
**messages_params: object,
) -> tuple[Mapping[str, object], Mapping[str, object]]:
route: Final = respx_mock.post(url__regex=upstream_url).mock(return_value=httpx.Response(200, json=upstream_body))
deployment: Final = {"model_name": "group", "litellm_params": {**litellm_params}, "model_info": {}}
sibling: Final = {"model_name": "sibling", "litellm_params": {**litellm_params}, "model_info": {**sibling_model_info}}
await Router(model_list=[deployment]).aanthropic_messages(
model="group", messages=MESSAGES, max_tokens=4096, **messages_params
)
alone: Final = json.loads(route.calls.last.request.read())
await Router(model_list=[sibling, deployment]).aanthropic_messages(
model="group", messages=MESSAGES, max_tokens=4096, **messages_params
)
return alone, json.loads(route.calls.last.request.read())
class TestASiblingDeploymentLeavesTheOffSwitchAlone:
@pytest.fixture(autouse=True)
def _httpx_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"litellm_params, api_base",
[
({**_OFF_MAP_OPENAI_DEPLOYMENT, "model": "openai/vendor/off-map-hybrid-reasoner"}, r"https://compatible\.example\.test/v1"),
({"model": "openai/gpt-5-mini", "api_key": "fake-key"}, r"https://api\.openai\.com/v1"),
],
ids=["off-map", "catalog"],
)
@pytest.mark.parametrize(
"uses_chat_route, reasoning_key", [(True, "reasoning_effort"), (False, "reasoning")], ids=["chat-route", "responses-bridge"]
)
async def test_a_first_turn_between_tools_sends_what_the_deployment_sends_alone(
self,
respx_mock: respx.MockRouter,
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
litellm_params: Mapping[str, object],
api_base: str,
uses_chat_route: bool,
reasoning_key: str,
) -> None:
upstream_url: Final = f"{api_base}/{'chat/completions' if uses_chat_route else 'responses'}"
upstream_body: Final = _CHAT_COMPLETION if uses_chat_route else _RESPONSE
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", uses_chat_route)
alone, beside = await _bodies_sent_alone_and_beside_a_sibling(
respx_mock,
litellm_params,
{"supports_none_reasoning_effort": True},
upstream_url,
upstream_body,
thinking={"type": "between_tools"},
)
assert beside.get(reasoning_key) == alone.get(reasoning_key)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"messages_params",
[{"thinking": "enabled"}, {"thinking": "", "output_config": {"effort": "low"}}],
ids=["truthy", "falsy-with-effort"],
)
async def test_a_non_object_thinking_still_reaches_the_responses_bridge(
respx_mock: respx.MockRouter,
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
messages_params: Mapping[str, object],
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
sent: Final = await _body_sent_upstream(
respx_mock,
{"model": "openai/gpt-5-mini", "api_key": "fake-key"},
{},
r"https://api\.openai\.com/v1/responses",
_RESPONSE,
**messages_params,
)
assert sent["model"] == "gpt-5-mini"
_CLIENT_EFFORT_NONE: Final = [
{"thinking": {"type": "adaptive"}, "output_config": {"effort": "none"}},
{"output_config": {"effort": "none"}},
]
class TestAClientEffortNoneFollowsTheUnsupportedValueRule:
@pytest.fixture(autouse=True)
def _httpx_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
@pytest.mark.asyncio
@pytest.mark.parametrize("messages_params", _CLIENT_EFFORT_NONE, ids=["adaptive", "bare"])
@pytest.mark.parametrize(
"uses_chat_route, reasoning_key", [(True, "reasoning_effort"), (False, "reasoning")], ids=["chat-route", "responses-bridge"]
)
async def test_drop_params_drops_it_on_a_model_that_refuses_none(
self,
respx_mock: respx.MockRouter,
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
messages_params: Mapping[str, object],
uses_chat_route: bool,
reasoning_key: str,
) -> None:
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", uses_chat_route)
sent: Final = await _body_sent_upstream(
respx_mock,
{"model": "openai/gpt-5-mini", "api_key": "fake-key", "drop_params": True},
{},
rf"https://api\.openai\.com/v1/{'chat/completions' if uses_chat_route else 'responses'}",
_CHAT_COMPLETION if uses_chat_route else _RESPONSE,
**messages_params,
)
assert reasoning_key not in sent
@pytest.mark.asyncio
@pytest.mark.parametrize("messages_params", _CLIENT_EFFORT_NONE, ids=["adaptive", "bare"])
@pytest.mark.parametrize("uses_chat_route", [True, False], ids=["chat-route", "responses-bridge"])
async def test_without_drop_params_it_answers_400(
self,
respx_mock: respx.MockRouter,
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
messages_params: Mapping[str, object],
uses_chat_route: bool,
) -> None:
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", uses_chat_route)
with pytest.raises(litellm.UnsupportedParamsError, match="effort=none is not supported for gpt-5-mini"):
await _body_sent_upstream(
respx_mock,
{"model": "openai/gpt-5-mini", "api_key": "fake-key"},
{},
rf"https://api\.openai\.com/v1/{'chat/completions' if uses_chat_route else 'responses'}",
_CHAT_COMPLETION if uses_chat_route else _RESPONSE,
**messages_params,
)
class TestTheOffSwitchUsesMinimalOnlyWhereTheDeploymentConfirmsIt:
@pytest.fixture(autouse=True)
def _httpx_transport(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
@pytest.mark.asyncio
@pytest.mark.parametrize(
"model_info, expected",
[
({"supports_none_reasoning_effort": False}, "low"),
({"reasoning_effort_levels": ["minimal", "low", "medium", "high"]}, "minimal"),
],
ids=["none-flag-only", "declared-levels"],
)
@pytest.mark.parametrize("thinking", [{"type": "disabled"}, {"type": "between_tools"}], ids=["disabled", "between_tools"])
@pytest.mark.parametrize("uses_chat_route", [True, False], ids=["chat-route", "responses-bridge"])
@pytest.mark.parametrize("drop_params", [False, True], ids=["no-drop_params", "drop_params"])
async def test_an_o3_deployment_gets_the_lowest_level_it_confirms(
self,
respx_mock: respx.MockRouter,
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
model_info: Mapping[str, object],
expected: str,
thinking: Mapping[str, object],
uses_chat_route: bool,
drop_params: bool,
) -> None:
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", uses_chat_route)
sent: Final = await _body_sent_upstream(
respx_mock,
{"model": "openai/o3", "api_key": "fake-key", "drop_params": drop_params},
model_info,
rf"https://api\.openai\.com/v1/{'chat/completions' if uses_chat_route else 'responses'}",
_CHAT_COMPLETION if uses_chat_route else _RESPONSE,
thinking=thinking,
)
assert (sent["reasoning_effort"] if uses_chat_route else sent["reasoning"]["effort"]) == expected

303
tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_handler.py
View file

@ -4,9 +4,10 @@ import os
import sys
from collections.abc import AsyncIterator, Mapping
from itertools import chain
from typing import Final, cast
from typing import Final, Literal, cast
from unittest.mock import AsyncMock, patch
import httpx
import pytest
import respx
@ -21,8 +22,11 @@ from litellm.llms.anthropic.messages.responses_adapters.handler import (
LiteLLMMessagesToResponsesAPIHandler,
_build_responses_kwargs,
)
from litellm.llms.anthropic.messages.utils import model_reasoning_efforts
from litellm.router import Router
from litellm.litellm_core_utils.reasoning_effort_capability import resolve_supported_reasoning_efforts
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS, converted_stream_requested
from litellm.types.llms.anthropic import AnthropicOutputConfig
MESSAGES = [{"role": "user", "content": "hello"}]
CLAUDE_CODE_USER_ID = json.dumps({"device_id": "d" * 64, "account_uuid": "", "session_id": "session-abc"})
@ -132,6 +136,184 @@ def test_build_responses_kwargs_auto_summary_asks_each_target_for_a_level_it_acc
assert cast(Mapping[str, object], responses_kwargs["reasoning"])["summary"] == summary
OFF_SWITCH_MODELS: Final = [
"gpt-5",
"gpt-5-mini",
"gpt-5-nano",
"gpt-6.1-sol",
"gpt-6-luna",
"gpt-5.1",
"chat-latest",
"o3",
"gpt-4.1",
"gpt-bridge-unmapped",
]
OFF_THINKING: Final = [{"type": "disabled"}, {"type": "between_tools"}]
OFF_THINKING_IDS: Final = ["disabled", "between_tools-first-turn"]
def _leaves_reasoning_to_the_model(model: str, thinking: dict[str, object]) -> bool:
return thinking["type"] == "between_tools" and model_reasoning_efforts(model, "openai") is None
def _lowest_listed_off_switch(model: str) -> str:
levels: Final = model_reasoning_efforts(model, "openai") or ()
return next((level for level in ("none", "minimal", "low") if level in levels), "none")
@pytest.mark.parametrize("thinking", OFF_THINKING, ids=OFF_THINKING_IDS)
@pytest.mark.parametrize("model", OFF_SWITCH_MODELS)
def test_build_responses_kwargs_turns_reasoning_off_at_the_lowest_level_the_model_lists_unless_it_never_reasons(
model: str, thinking: dict[str, object]
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model=f"openai/{model}",
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai"},
)
map_row: Final = litellm.model_cost.get(model)
never_reasons: Final = map_row is not None and not map_row.get("supports_reasoning")
sends_no_reasoning: Final = never_reasons or _leaves_reasoning_to_the_model(model, thinking)
assert responses_kwargs.get("reasoning") == (
None if sends_no_reasoning else {"effort": _lowest_listed_off_switch(model)}
)
@pytest.mark.parametrize("thinking", OFF_THINKING, ids=OFF_THINKING_IDS)
@pytest.mark.parametrize("model", OFF_SWITCH_MODELS)
def test_build_responses_kwargs_under_drop_params_keeps_a_listed_lower_tier_and_drops_an_unconfirmed_none(
model: str, thinking: dict[str, object]
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model=f"openai/{model}",
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai", "drop_params": True},
)
off_switch: Final = _lowest_listed_off_switch(model)
map_row: Final = litellm.model_cost.get(model)
confirms_none: Final = map_row is not None and map_row.get("supports_none_reasoning_effort") is True
keeps_none: Final = map_row is None or (bool(map_row.get("supports_reasoning")) and confirms_none)
sends_off_switch: Final = (off_switch != "none" or keeps_none) and not _leaves_reasoning_to_the_model(
model, thinking
)
assert responses_kwargs.get("reasoning") == ({"effort": off_switch} if sends_off_switch else None)
@pytest.mark.parametrize("model", ["gpt-5.4-mini", "gpt-5.1", "gpt-6-luna", "gpt-bridge-unmapped"])
@pytest.mark.parametrize("thinking", [None, {"type": "adaptive"}], ids=["thinking-left-out", "adaptive"])
def test_build_responses_kwargs_never_turns_on_reasoning_the_client_left_off(
model: str, thinking: dict[str, object] | None
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model=f"openai/{model}",
output_config={"effort": "high"},
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai"},
)
map_row: Final = litellm.model_cost.get(model)
reasons_unasked: Final = map_row is not None and map_row.get("default_reasoning_effort") != "none"
reasoning: Final = responses_kwargs.get("reasoning")
assert (reasoning is None) == (thinking is None and not reasons_unasked)
assert reasoning is None or cast(Mapping[str, object], reasoning)["effort"] != "none"
@pytest.mark.parametrize("thinking", [None, {"type": "adaptive"}], ids=["thinking-left-out", "adaptive"])
def test_build_responses_kwargs_leaves_a_bare_output_effort_to_an_explicit_reasoning_effort(
thinking: dict[str, object] | None,
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model="openai/gpt-5-mini",
output_config={"effort": "high"},
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai", "reasoning_effort": "minimal"},
)
assert responses_kwargs["reasoning_effort"] == "minimal"
assert ("reasoning" in responses_kwargs) == (thinking is not None)
@pytest.mark.parametrize("drop_params", [True, False], ids=["drop_params", "no-drop_params"])
@pytest.mark.parametrize("thinking", [None, {"type": "adaptive"}], ids=["thinking-left-out", "adaptive"])
def test_build_responses_kwargs_drops_an_effort_no_reasoning_api_defines_only_under_drop_params(
drop_params: bool, thinking: dict[str, object] | None
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model="openai/gpt-6-luna",
output_config=cast(AnthropicOutputConfig, {"effort": "turbo"}),
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai", "drop_params": drop_params},
)
assert responses_kwargs.get("reasoning") == (None if drop_params else {"effort": "turbo"})
@pytest.mark.parametrize("drop_params", [True, False], ids=["drop_params", "no-drop_params"])
@pytest.mark.parametrize("thinking", [None, {"type": "adaptive"}], ids=["thinking-left-out", "adaptive"])
@pytest.mark.parametrize("effort", ["low", "medium", "high"])
def test_build_responses_kwargs_drops_an_effort_the_model_does_not_list_only_under_drop_params(
effort: Literal["low", "medium", "high"], thinking: dict[str, object] | None, drop_params: bool
) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model="openai/chat-latest",
output_config={"effort": effort},
thinking=thinking,
extra_kwargs={"custom_llm_provider": "openai", "drop_params": drop_params},
)
listed: Final = effort in (model_reasoning_efforts("chat-latest", "openai") or ())
assert responses_kwargs.get("reasoning") == ({"effort": effort} if listed or not drop_params else None)
def test_build_responses_kwargs_under_drop_params_leaves_an_unlisted_effort_to_a_deployment_that_rounds_it() -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model="openai/chat-latest",
output_config={"effort": "high"},
thinking={"type": "adaptive"},
extra_kwargs={
"custom_llm_provider": "openai",
"drop_params": True,
"model_info": {"reasoning_effort_levels": ["low", "high"]},
},
)
assert responses_kwargs.get("reasoning") == {"effort": "high"}
@pytest.mark.parametrize("effort", ["max", "xhigh"])
def test_build_responses_kwargs_forwards_an_output_effort_the_model_accepts(effort: Literal["max", "xhigh"]) -> None:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=1024,
messages=MESSAGES,
model="openai/gpt-5.6-luna",
output_config={"effort": effort},
extra_kwargs={"custom_llm_provider": "openai"},
)
forwarded: Final = cast(Mapping[str, object], responses_kwargs["reasoning"])["effort"]
accepted: Final = resolve_supported_reasoning_efforts(
litellm.get_model_info(model="gpt-5.6-luna", custom_llm_provider="openai"), deployment_is_mapped=True
)
assert accepted is not None and forwarded in accepted
assert (forwarded == effort) == (effort in accepted)
def test_build_responses_kwargs_keeps_the_deployment_include_next_to_encrypted_reasoning():
responses_kwargs = _build_responses_kwargs(
max_tokens=1024,
@ -1135,3 +1317,122 @@ async def test_the_sync_path_reports_its_edits_on_the_streamed_message_delta(
]
message_delta: Final = next(event for event in events if event["type"] == "message_delta")
assert message_delta["context_management"]["applied_edits"][0]["cleared_tool_uses"] == 2
SUMMARY_REFUSAL: Final = {
"error": {
"message": (
"Your organization must be verified to generate reasoning summaries. Please go to: "
"https://platform.openai.com/settings/organization/general and click on Verify Organization. "
"If you just verified, it can take up to 15 minutes for access to propagate."
),
"type": "invalid_request_error",
"param": "reasoning.summary",
"code": "unsupported_value",
}
}
OTHER_SUMMARY_400: Final = {
"error": {
"message": "Unsupported value: 'reasoning.summary' does not support 'auto' with this model.",
"type": "invalid_request_error",
"param": "reasoning.summary",
"code": "unsupported_value",
}
}
def _upstream_answering_400_then_200(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, *, refusal: Mapping[str, object], stream: bool
) -> respx.Route:
monkeypatch.setenv("OPENAI_API_KEY", "sk-summary-refusal-test")
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
litellm.in_memory_llm_clients_cache.flush_cache()
served: Final = (
httpx.Response(200, headers={"Content-Type": "text/event-stream"}, content=RESPONSES_SSE_BODY)
if stream
else httpx.Response(200, json=TEXT_RESPONSE)
)
return respx_mock.post("https://api.openai.com/v1/responses").mock(
side_effect=[httpx.Response(400, json=refusal), served]
)
async def _summarized_turn(*, sync: bool, stream: bool, drop_params: bool) -> object:
if sync:
return LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
max_tokens=1024,
messages=MESSAGES,
model="openai/gpt-5.6-luna",
stream=stream,
custom_llm_provider="openai",
thinking={"type": "adaptive", "display": "summarized"},
output_config={"effort": "high"},
drop_params=drop_params,
)
return await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
max_tokens=1024,
messages=MESSAGES,
model="openai/gpt-5.6-luna",
stream=stream,
custom_llm_provider="openai",
thinking={"type": "adaptive", "display": "summarized"},
output_config={"effort": "high"},
drop_params=drop_params,
)
async def _reply_types(served: object, *, stream: bool) -> list[object]:
if not stream:
return [cast(Mapping[str, object], served)["type"]]
return [
json.loads(chunk.decode().split("data: ", 1)[1])["type"]
async for chunk in cast(AsyncIterator[bytes], served)
]
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True], ids=["non_stream", "stream"])
@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"])
async def test_under_drop_params_an_unverified_organization_gets_the_turn_again_without_the_summary(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sync: bool, stream: bool
) -> None:
upstream: Final = _upstream_answering_400_then_200(respx_mock, monkeypatch, refusal=SUMMARY_REFUSAL, stream=stream)
reply_types: Final = await _reply_types(
await _summarized_turn(sync=sync, stream=stream, drop_params=True), stream=stream
)
first, second = (json.loads(call.request.content) for call in upstream.calls)
assert first["reasoning"]["summary"]
assert second["reasoning"] == {"effort": first["reasoning"]["effort"]}
assert {key: value for key, value in second.items() if key != "reasoning"} == {
key: value for key, value in first.items() if key != "reasoning"
}
assert (reply_types[0], reply_types[-1]) == (("message_start", "message_stop") if stream else ("message",) * 2)
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True], ids=["non_stream", "stream"])
@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"])
async def test_without_drop_params_an_unverified_organization_gets_the_400_after_one_request(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sync: bool, stream: bool
) -> None:
upstream: Final = _upstream_answering_400_then_200(respx_mock, monkeypatch, refusal=SUMMARY_REFUSAL, stream=stream)
with pytest.raises(litellm.BadRequestError, match="must be verified to generate reasoning summaries"):
await _reply_types(await _summarized_turn(sync=sync, stream=stream, drop_params=False), stream=stream)
assert upstream.call_count == 1
@pytest.mark.asyncio
@pytest.mark.parametrize("sync", [False, True], ids=["async", "sync"])
async def test_under_drop_params_another_400_about_the_summary_is_not_retried(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sync: bool
) -> None:
upstream: Final = _upstream_answering_400_then_200(respx_mock, monkeypatch, refusal=OTHER_SUMMARY_400, stream=False)
with pytest.raises(litellm.BadRequestError, match="does not support 'auto' with this model"):
await _summarized_turn(sync=sync, stream=False, drop_params=True)
assert upstream.call_count == 1

164
tests/unit/llms/anthropic/messages/responses_adapters/test_responses_adapters_transformation.py
View file

@ -12,6 +12,7 @@ import pytest
from openai.types.responses import ResponseInputImageParam
from pydantic import TypeAdapter
import litellm
from litellm.constants import (
DEFAULT_REASONING_EFFORT_HIGH_THINKING_BUDGET,
DEFAULT_REASONING_EFFORT_LOW_THINKING_BUDGET,
@ -29,6 +30,7 @@ from litellm.llms.openai.responses.transformation import OpenAIResponsesAPIConfi
from litellm.types.llms.anthropic import (
AllAnthropicToolsValues,
AnthropicMessagesRequest,
AnthropicOutputConfig,
)
from litellm.types.llms.openai import ResponseAPIUsage
@ -1027,10 +1029,6 @@ class TestTranslateThinkingToReasoning:
assert result_low["effort"] == "low"
assert "summary" not in result_low
def test_disabled_type_returns_none(self):
result = _ADAPTER.translate_thinking_to_reasoning({"type": "disabled"}, model=_MODEL)
assert result is None
def test_non_dict_returns_none(self):
result = _ADAPTER.translate_thinking_to_reasoning("enabled", model=_MODEL) # type: ignore
assert result is None
@ -1074,6 +1072,159 @@ class TestTranslateThinkingToReasoning:
os.environ.pop("LITELLM_REASONING_AUTO_SUMMARY", None)
def _set_auto_summary(monkeypatch: pytest.MonkeyPatch, enabled: bool) -> None:
monkeypatch.setattr(litellm, "reasoning_auto_summary", enabled)
monkeypatch.delenv("LITELLM_REASONING_AUTO_SUMMARY", raising=False)
_FRESH_TURN: Final = ({"role": "user", "content": "What is the weather in Denver?"},)
_TOOL_CALL_TURNS: Final = (
*_FRESH_TURN,
{
"role": "assistant",
"content": [{"type": "tool_use", "id": "toolu_1", "name": "weather", "input": {"city": "Denver"}}],
},
{
"role": "user",
"content": [
{"type": "tool_result", "tool_use_id": "toolu_1", "content": "Sunny"},
{"type": "text", "text": "Keep it short."},
],
},
)
_NEW_QUESTION_AFTER_TOOL_CALLS: Final = (
*_TOOL_CALL_TURNS,
{"role": "assistant", "content": "It is sunny."},
{"role": "user", "content": [{"type": "text", "text": "And tomorrow?"}]},
)
_TOOL_RESULT_SPLIT_FROM_TEXT: Final = (
*_TOOL_CALL_TURNS[:2],
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "toolu_1", "content": "Sunny"}]},
{"role": "user", "content": "Keep it short."},
)
_LOW_BUDGET: Final = {"type": "enabled", "budget_tokens": DEFAULT_REASONING_EFFORT_LOW_THINKING_BUDGET}
@pytest.mark.parametrize(
"thinking, output_config, messages, reasoning",
[
pytest.param({"type": "disabled"}, {"effort": "high"}, _FRESH_TURN, {"effort": "none"}, id="disabled"),
pytest.param(
{"type": "between_tools"}, {"effort": "high"}, _FRESH_TURN, {"effort": "none"}, id="between-fresh"
),
pytest.param(
{"type": "between_tools"},
{"effort": "high"},
_NEW_QUESTION_AFTER_TOOL_CALLS,
{"effort": "none"},
id="between-new-question-after-tool-calls",
),
pytest.param(
{"type": "between_tools"},
{"effort": "high"},
_TOOL_CALL_TURNS,
{"effort": "high", "summary": "auto"},
id="between-tool-result",
),
pytest.param(
{"type": "between_tools"},
{"effort": "high"},
_TOOL_RESULT_SPLIT_FROM_TEXT,
{"effort": "high", "summary": "auto"},
id="between-tool-result-split-from-text",
),
pytest.param(
{"type": "between_tools"},
None,
_TOOL_CALL_TURNS,
{"effort": "medium", "summary": "auto"},
id="between-default",
),
pytest.param(
{"type": "between_tools", "summary": "concise"},
{"effort": "high"},
_TOOL_CALL_TURNS,
{"effort": "high", "summary": "concise"},
id="between-tool-result-own-summary",
),
pytest.param(
{"type": "between_tools", "display": "omitted"},
{"effort": "high"},
_TOOL_CALL_TURNS,
{"effort": "high"},
id="between-tool-result-omitted",
),
pytest.param(None, {"effort": "low"}, _FRESH_TURN, {"effort": "low"}, id="output-effort-alone"),
pytest.param({"type": "future"}, {"effort": "low"}, _FRESH_TURN, {"effort": "low"}, id="unknown-type-effort"),
pytest.param({"type": "adaptive"}, {"effort": "xhigh"}, _FRESH_TURN, {"effort": "xhigh"}, id="adaptive-effort"),
pytest.param({"type": "adaptive"}, None, _FRESH_TURN, {"effort": "medium"}, id="adaptive-default"),
pytest.param(_LOW_BUDGET, {"effort": "high"}, _FRESH_TURN, {"effort": "low"}, id="enabled-budget-wins"),
pytest.param(None, None, _FRESH_TURN, None, id="nothing-set"),
pytest.param({"type": "future"}, None, _FRESH_TURN, None, id="unknown-type-alone"),
],
)
def test_translate_request_turns_thinking_and_output_effort_into_reasoning(
monkeypatch: pytest.MonkeyPatch,
thinking: dict[str, object] | None,
output_config: AnthropicOutputConfig | None,
messages: tuple[dict[str, object], ...],
reasoning: dict[str, object] | None,
) -> None:
_set_auto_summary(monkeypatch, False)
request: Final = _make_request(messages=list(messages), thinking=thinking, output_config=output_config)
assert _ADAPTER.translate_request(request).get("reasoning") == reasoning
@pytest.mark.parametrize(
"auto_summary, thinking, output_config, reasoning",
[
pytest.param(
False,
{"type": "adaptive", "display": "summarized"},
None,
{"effort": "medium", "summary": "auto"},
id="summarized",
),
pytest.param(
False,
{"type": "adaptive", "display": "updates"},
None,
{"effort": "medium", "summary": "auto"},
id="updates",
),
pytest.param(
False, {**_LOW_BUDGET, "display": "summarized"}, None, {"effort": "low", "summary": "auto"}, id="enabled"
),
pytest.param(True, {"type": "adaptive", "display": "omitted"}, None, {"effort": "medium"}, id="omitted"),
pytest.param(
True,
{"type": "adaptive", "summary": "detailed", "display": "summarized"},
None,
{"effort": "medium", "summary": "detailed"},
id="summary-beats-display",
),
pytest.param(True, {"type": "adaptive"}, None, {"effort": "medium", "summary": "concise"}, id="auto-summary"),
pytest.param(False, {"type": "adaptive"}, None, {"effort": "medium"}, id="no-summary-asked"),
pytest.param(True, None, {"effort": "low"}, {"effort": "low", "summary": "concise"}, id="output-effort-alone"),
pytest.param(True, {"type": "disabled", "display": "summarized"}, None, {"effort": "none"}, id="disabled"),
pytest.param(
True, {"type": "between_tools", "display": "summarized"}, None, {"effort": "none"}, id="between-fresh"
),
],
)
def test_translate_thinking_to_reasoning_asks_for_the_summary_the_client_shows(
monkeypatch: pytest.MonkeyPatch,
auto_summary: bool,
thinking: dict[str, object] | None,
output_config: dict[str, object] | None,
reasoning: dict[str, object],
) -> None:
_set_auto_summary(monkeypatch, auto_summary)
assert _ADAPTER.translate_thinking_to_reasoning(thinking, output_config, auto_summary_level="concise", model=_MODEL) == reasoning
# ---------------------------------------------------------------------------
# translate_request – broader coverage
# ---------------------------------------------------------------------------
@ -1185,11 +1336,6 @@ class TestTranslateRequestBroaderCoverage:
assert kwargs["reasoning"] == {"effort": "high"}
assert "summary" not in kwargs["reasoning"]
def test_disabled_thinking_not_included_in_kwargs(self):
req = _make_request(thinking={"type": "disabled"})
kwargs = _ADAPTER.translate_request(req)
assert "reasoning" not in kwargs
def test_thinking_asks_for_the_encrypted_reasoning(self):
"""The documented way to get reasoning that survives store=false is to ask for it."""
req = _make_request(thinking={"type": "enabled", "budget_tokens": 12000})

47
tests/unit/llms/anthropic/messages/test_reasoning_effort_fields.py
View file

@ -18,7 +18,9 @@ from litellm.litellm_core_utils.reasoning_effort_capability import (
resolve_supported_reasoning_efforts,
)
from litellm.llms.anthropic.messages.utils import (
maps_bare_output_effort,
normalize_reasoning_effort_value,
requested_reasoning_summary,
)
from litellm.utils import get_model_info
@ -318,3 +320,48 @@ class TestADeclarationDisjointFromTheChain:
what every deployment got before the resolver was consulted. Dropping the parameter outright
is the real answer and belongs with the callers that build the request."""
assert normalize_reasoning_effort_value(effort, declared_effort_entry, "synthetic") == expected
@pytest.fixture
def synthetic_model_entry(local_model_cost_map, request):
key = f"synthetic/{request.node.name}"
litellm.model_cost[key] = {"litellm_provider": "synthetic", "mode": "chat", **request.param}
litellm.get_model_info.cache_clear()
try:
yield key.removeprefix("synthetic/")
finally:
litellm.model_cost.pop(key, None)
litellm.get_model_info.cache_clear()
@pytest.mark.parametrize(
"synthetic_model_entry, maps",
[
({"supports_reasoning": True}, True),
({"supports_reasoning": True, "default_reasoning_effort": "none"}, False),
({"supports_reasoning": False}, False),
({}, False),
],
indirect=["synthetic_model_entry"],
ids=["reasons-by-default", "reasoning-off-by-default", "cannot-reason", "no-reasoning-flag"],
)
def test_a_bare_output_effort_maps_only_onto_a_model_that_reasons_by_default(synthetic_model_entry, maps):
assert maps_bare_output_effort(synthetic_model_entry, "synthetic", deployment_model_info=None) is maps
@pytest.mark.parametrize(
"thinking, summary",
[
({"type": "disabled", "display": "summarized", "summary": "detailed"}, None),
({"type": "adaptive", "display": "omitted", "summary": "concise"}, "concise"),
({"type": "adaptive", "display": "omitted"}, None),
({"type": "adaptive", "display": "summarized"}, "auto"),
({"type": "adaptive"}, None),
({"type": "between_tools"}, "auto"),
({"type": "between_tools", "display": "omitted"}, None),
],
)
def test_requested_reasoning_summary_takes_the_client_summary_then_the_display_and_never_one_for_disabled(
thinking: dict[str, object], summary: str | None
) -> None:
assert requested_reasoning_summary(thinking) == summary