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

fix(chatgpt): round-trip encrypted reasoning through /v1/messages (#49076)

Browse source 
* fix(chatgpt): round-trip encrypted reasoning through /v1/messages

The chat bridge stored encrypted reasoning on reasoning_items, the Anthropic
adapter never read that field, stream chunks that only carried thinking
blocks were dropped, and the messages adapter stripped the signed blocks
before a chatgpt request could turn them back into reasoning items.

Co-authored-by: DDT <1786035110@qq.com>

* fix(anthropic): keep Responses reasoning on tool turns and retry once when a backend cannot decrypt it

* fix(anthropic): stop logging rejected encrypted reasoning

Quote the retry helper return type so importing litellm no longer evaluates a string alias with |. The warning keeps provider, model, status, and error code, and leaves the ciphertext out.

* fix(anthropic): keep encrypted reasoning on OpenAI thinking and xAI web search turns, and stream signatures only to Messages clients

The keep check now knows when the handler moves an OpenAI thinking turn to Responses, and when web_search_options sends an xAI turn there.

The Responses-to-chat stream iterator adds the signed reasoning chunk only when the caller is /v1/messages, so /v1/chat/completions streams stay unchanged.

The reasoning retry fires only when the translated request still carries encrypted reasoning.

* fix(anthropic): keep each streamed reasoning summary with its own item, and read the Messages marker before the bridge call

* fix(anthropic): send openai/ Messages turns again once without encrypted reasoning a backend rejects

The openai/ Responses path on /v1/messages had no retry, so a chatgpt/ block left in history answered 400 on every later openai/ turn. That path now sends the turn once more without encrypted reasoning, the same way the chat bridge does. The retry matches only invalid_encrypted_content or a "could not be verified" message naming the content, so the exception mapper's own hint no longer triggers it. The xAI web search keep path is backed out, since xAI sends no encrypted reasoning over the bridge.

* fix(anthropic): keep every non-streamed reasoning item for Messages clients and read a deployment's reasoning_summary in the keep check

* fix(anthropic): log only the outcome of a turn sent again without rejected reasoning

A /v1/messages turn whose replayed reasoning the backend rejects is sent once more without it. The rejected first attempt ran the failure callbacks on the shared logging object, which logged a failure for a turn that then succeeded, kept the retry's own failure from being logged, and freed the parallel-request slot while the retry was still in flight.

The first attempt now runs inside resent_on, and the client wrapper skips failure logging and the budget unbind for a failure that matches it. The retry runs outside that context, so its own outcome logs as usual.

* test(anthropic): type the parameters of the failure-recording test loggers

* fix(anthropic): skip the SDK num_retries retry for an attempt that is sent again

* fix(anthropic): stop SDK retries from resending a rejected block or clearing the global num_retries

A transient 429 or 500 on a /v1/messages turn that carries a replayed encrypted
block was retried by an SDK call's num_retries loop, which sent the block again
until every retry was rejected. On async calls the loop then raised the first
transient error, so the turn failed instead of being sent again without the
block. The retry loop now stops at that rejection and passes it on, so the turn
is sent again once without encrypted blocks.

A rejected first try on a sync call also cleared the process-wide
litellm.num_retries, so later calls in the process were no longer retried. That
setting now stays as configured.

* fix(anthropic): stop a cancelled SDK retry from starting another call

The resent-failure retry predicate returned true for BaseException
subclasses such as asyncio.CancelledError, so a cancelled
acompletion_with_retries or aresponses_with_retries call could start
another provider call. Retry only Exception subclasses again, as
tenacity's default did, while still skipping a resent failure.

* test(anthropic): type the retry helper parameter in the cancellation test

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
DDT 2026-11-08 03:32:59 +08:00 • committed by GitHub
parent cdbb487926
commit d0c16008c8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 2191 additions and 183 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
litellm/_internal_context.py
View file

@ -44,6 +44,8 @@ _service_caller: Final[ContextVar[str | None]] = ContextVar("service_caller", de
_mirrored_call: Final[ContextVar[bool]] = ContextVar("mirrored_call", default=False)
_resent_on: Final[ContextVar[Callable[[Exception], bool] | None]] = ContextVar("resent_on", default=None)
_request_parent_otel_span: Final[ContextVar[Span | None]] = ContextVar("request_parent_otel_span", default=None)
@ -168,6 +170,23 @@ def in_mirrored_call() -> bool:
return _mirrored_call.get()
@contextmanager
def resent_on(rejection: Callable[[Exception], bool]) -> Generator[None]:
"""An attempt its caller sends again when it fails with ``rejection``. That failure is not the request's
outcome, so it is neither logged as the request's failure nor retried, and its budget reservation and rate-limit
slot stay with the attempt sent again."""
token: Final = _resent_on.set(rejection)
try:
yield
finally:
_restore(_resent_on, token)
def is_resent_failure(error: Exception) -> bool:
rejection: Final = _resent_on.get()
return rejection is not None and rejection(error)
def set_request_parent_otel_span(span: Span) -> None:
"""Anchor the proxy request's OTEL span for datastore service spans that no call site passes a parent to.
Never reset: it dies with the request task and rides the tasks it spawns, like ``request.state``."""

8
litellm/completion_extras/litellm_responses_transformation/handler.py
View file

@ -182,6 +182,9 @@ class ResponsesToCompletionBridgeHandler:
model_response: Final = validated_kwargs["model_response"]
logging_obj: Final = validated_kwargs["logging_obj"]
custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
from .transformation import serves_anthropic_messages
streams_reasoning_signatures: Final = serves_anthropic_messages(logging_obj)
if kwargs.get("stream") is True and "stream" not in optional_params:
optional_params = {**optional_params, "stream": True}
logging_obj.record_bridged_chat_request(optional_params, litellm_params)
@ -262,6 +265,7 @@ class ResponsesToCompletionBridgeHandler:
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
replays_openai_history=custom_llm_provider == "openai",
streams_reasoning_signatures=streams_reasoning_signatures,
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,
@ -284,6 +288,9 @@ class ResponsesToCompletionBridgeHandler:
model_response: Final = validated_kwargs["model_response"]
logging_obj: Final = validated_kwargs["logging_obj"]
custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
from .transformation import serves_anthropic_messages
streams_reasoning_signatures: Final = serves_anthropic_messages(logging_obj)
if kwargs.get("stream") is True and "stream" not in optional_params:
optional_params = {**optional_params, "stream": True}
logging_obj.record_bridged_chat_request(optional_params, litellm_params)
@ -367,6 +374,7 @@ class ResponsesToCompletionBridgeHandler:
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
replays_openai_history=custom_llm_provider == "openai",
streams_reasoning_signatures=streams_reasoning_signatures,
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,

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

@ -39,7 +39,12 @@ from litellm.litellm_core_utils.allowed_tools_tool_choice import (
parse_chat_allowed_tools_tool_choice,
responses_allowed_tools_tool_choice,
)
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params, iter_items
from litellm.litellm_core_utils.core_helpers import (
as_str_mapping,
attribute_or_none,
drops_unsupported_params,
iter_items,
)
from litellm.litellm_core_utils.hidden_params import get_hidden_params, get_or_create_hidden_params
from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils import (
prompt_cache_breakpoint_with_readable_mode,
@ -48,6 +53,7 @@ from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils i
)
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
responses_reasoning_items_from_thinking_blocks,
thinking_blocks_from_reasoning_items,
)
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
from litellm.llms.base_llm.bridges.completion_transformation import (
@ -69,6 +75,8 @@ from litellm.responses.utils import (
from litellm.types.llms.openai import (
ChatCompletionAnnotation,
ChatCompletionReasoningItem,
ChatCompletionRedactedThinkingBlock,
ChatCompletionThinkingBlock,
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolParamFunctionChunk,
@ -77,7 +85,7 @@ from litellm.types.llms.openai import (
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
from litellm.types.utils import CallTypes, Delta, GenericStreamingChunk, ModelResponseStream, StreamingChoices
if TYPE_CHECKING:
from openai.types.responses import ResponseInputImageParam, ResponseOutputItem
@ -93,8 +101,6 @@ if TYPE_CHECKING:
AllMessageValues,
ChatCompletionFileObject,
ChatCompletionImageObject,
ChatCompletionRedactedThinkingBlock,
ChatCompletionThinkingBlock,
ChatCompletionToolReferenceObject,
OpenAIMessageContentListBlock,
)
@ -264,6 +270,10 @@ class _BuiltReasoningItem(TypedDict):
summary: Sequence[_ReasoningSummaryText]
def serves_anthropic_messages(logging_obj: object) -> bool:
return attribute_or_none(logging_obj, "call_type") == CallTypes.anthropic_messages.value
def _get_reasoning_items(
msg: "AllMessageValues",
) -> list[ChatCompletionReasoningItem]:
@ -496,6 +506,77 @@ def _as_chat_reasoning_items(
return cast(list[ChatCompletionReasoningItem], list(reasoning_items))
def _with_stream_block_index(
block: Mapping[str, object],
index: int,
summary_streamed: bool,
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock:
if block.get("type") == "redacted_thinking":
data: Final = block.get("data")
return ChatCompletionRedactedThinkingBlock(
type="redacted_thinking",
data=data if isinstance(data, str) else "",
index=index,
)
thinking: Final = block.get("thinking")
signature: Final = block.get("signature")
return ChatCompletionThinkingBlock(
type="thinking",
thinking=thinking if isinstance(thinking, str) and not summary_streamed else "",
signature=signature if isinstance(signature, str) else None,
index=index,
)
def _stream_output_index(chunk: Mapping[str, object]) -> int | None:
output_index: Final = chunk.get("output_index")
return output_index if isinstance(output_index, int) and not isinstance(output_index, bool) else None
def _reasoning_done_thinking_blocks(
item: Mapping[str, object],
output_index: int | None,
summary_streamed: bool,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
index: Final = output_index if output_index is not None else 0
return tuple(
_with_stream_block_index(block, index, summary_streamed)
for block in thinking_blocks_from_reasoning_items((item,))
)
def _reasoning_done_stream_chunk(
chunk: Mapping[str, object],
summary_streamed_outputs: frozenset[int | None],
) -> ModelResponseStream | None:
event_type: Final = chunk.get("type")
event_name: Final = (
event_type.value
if isinstance(event_type, ResponsesAPIStreamEvents)
else event_type
if isinstance(event_type, str)
else None
)
if event_name != ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
return None
item: Final = as_str_mapping(chunk.get("item"))
if item is None or item.get("type") != "reasoning":
return None
output_index: Final = _stream_output_index(chunk)
signed_blocks: Final = _reasoning_done_thinking_blocks(item, output_index, output_index in summary_streamed_outputs)
if not signed_blocks:
return None
return ModelResponseStream(
choices=[
StreamingChoices(
index=0,
delta=Delta(thinking_blocks=list(signed_blocks)),
finish_reason=None,
)
]
)
def _shift_annotation(annotation: ChatCompletionAnnotation, offset: int) -> ChatCompletionAnnotation:
shifted: Final = {
key: value + offset if key in ("start_index", "end_index") and isinstance(value, int) else value
@ -1280,6 +1361,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
structured_output: bool = False,
replay_openai_history: bool = False,
completed: bool = False,
keeps_every_reasoning_item: bool = False,
) -> list["Choices"]:
"""
Convert Responses API output items to chat completion choices.
@ -1291,6 +1373,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
first message item becomes content
replay_openai_history: Record the output's item sequence for the next turn
completed: The response finished, so its item sequence is whole
keeps_every_reasoning_item: Keep each reasoning item before an answer, not only the last
Returns:
List of Choices objects
@ -1323,7 +1406,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
for item in output_items:
if isinstance(item, ResponseReasoningItem):
pending_reasoning_items = (
*(pending_reasoning_items if replay_openai_history else ()),
*(pending_reasoning_items if replay_openai_history or keeps_every_reasoning_item else ()),
_build_reasoning_item(
item_id=item.id,
encrypted_content=getattr(item, "encrypted_content", None),
@ -1573,6 +1656,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
)
== "openai",
completed=raw_response.status == "completed",
keeps_every_reasoning_item=serves_anthropic_messages(logging_obj),
)
response_is_incomplete: Final = raw_response.status == "incomplete" or (
@ -1637,6 +1721,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
json_mode: bool | None = False,
structured_output: bool = False,
replays_openai_history: bool = False,
streams_reasoning_signatures: bool = False,
) -> BaseModelResponseIterator:
return OpenAiResponsesToChatCompletionStreamIterator(
streaming_response,
@ -1644,6 +1729,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
json_mode,
structured_output=structured_output,
replays_openai_history=replays_openai_history,
streams_reasoning_signatures=streams_reasoning_signatures,
)
def _convert_content_str_to_input_text(self, content: str, role: str) -> dict[str, object]:
@ -2037,15 +2123,18 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
json_mode: bool | None = False,
structured_output: bool = False,
replays_openai_history: bool = False,
streams_reasoning_signatures: bool = False,
):
super().__init__(streaming_response, sync_stream, json_mode)
self._replays_openai_history: Final = replays_openai_history
self._streams_reasoning_signatures: Final = streams_reasoning_signatures
self._chat_completion_id: str | None = None
self._served_service_tier: str | None = None
self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state
self._structured_output: Final = structured_output
self._first_message_item_done: bool = False
self._reasoning_summary_streamed: bool = False
self._summary_streamed_outputs: frozenset[int | None] = frozenset()
def _handle_string_chunk(
self, str_line: Union[str, "BaseModel"]
@ -2263,7 +2352,6 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
)
]
)
elif event_type == "response.output_text.delta":
# Content part added to output
content_part = parsed_chunk.get("delta", None)
@ -2380,9 +2468,16 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
"""
verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk)
self._remember_served_service_tier(chunk)
reasoning_chunk: Final = (
_reasoning_done_stream_chunk(chunk, self._summary_streamed_outputs)
if self._streams_reasoning_signatures
else None
)
translated: Final = (
self._empty_content_chunk()
if self._is_text_delta_after_structured_answer(chunk)
else reasoning_chunk
if reasoning_chunk is not None
else OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
chunk,
tool_call_index_map=self._tool_call_index_map,
@ -2395,6 +2490,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
self._first_message_item_done = True
if _responses_event_type(chunk) == "response.reasoning_summary_text.delta" and chunk.get("delta"):
self._reasoning_summary_streamed = True
self._summary_streamed_outputs = self._summary_streamed_outputs | {_stream_output_index(chunk)}
return self._with_served_service_tier(self._with_stream_scoped_id(translated))
def _is_text_delta_after_structured_answer(self, chunk: Mapping[str, object]) -> bool:

95
litellm/litellm_core_utils/prompt_templates/reasoning_utils.py
View file

@ -5,6 +5,9 @@ from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from itertools import groupby
from typing import Final, cast # noqa: TID251 # Existing casts retained unchanged in this pure move
from pydantic import TypeAdapter
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.types.llms.openai import (
ChatCompletionReasoningItem,
ChatCompletionReasoningSummaryTextBlock,
@ -94,6 +97,90 @@ def encrypted_content_of_block(block: Mapping[str, object]) -> str | None:
return encrypted_content_from_signature(_encrypted_reasoning_field(block))
_REASONING_SUMMARY_SEPARATOR: Final = "\n\n"
_OBJECT_TUPLE: Final = TypeAdapter(tuple[object, ...])
def _object_sequence(value: object) -> tuple[object, ...]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
return ()
return _OBJECT_TUPLE.validate_python(value)
def _iter_mappings(value: object) -> tuple[Mapping[str, object], ...]:
return tuple(mapping for item in _object_sequence(value) if (mapping := as_str_mapping(item)) is not None)
def _reasoning_summary_part_text(part: object) -> str:
mapping: Final = as_str_mapping(part)
if mapping is not None:
text: Final = mapping.get("text")
return text if isinstance(text, str) else ""
text_attr: Final = getattr(part, "text", None)
return text_attr if isinstance(text_attr, str) else ""
def _reasoning_item_summary_text(item: Mapping[str, object]) -> str:
parts: Final = tuple(_reasoning_summary_part_text(part) for part in _object_sequence(item.get("summary")))
return _REASONING_SUMMARY_SEPARATOR.join(part for part in parts if part)
def thinking_block_from_reasoning_item(
item: object,
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock | None:
"""The chat thinking block that carries one Responses `reasoning` item back to an Anthropic client.
`encrypted_content` rides the same `litellm_encrypted_reasoning:` signature the Responses
adapter uses. A summary becomes thinking text. Bytes with no summary become `redacted_thinking`,
because an empty thinking string is not something the client can replay.
"""
mapping: Final = as_str_mapping(item)
if mapping is None or mapping.get("type") not in (None, "reasoning"):
return None
encrypted_content: Final = mapping.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content:
return None
signature: Final = encrypted_reasoning_signature(encrypted_content)
text: Final = _reasoning_item_summary_text(mapping)
if not text:
return ChatCompletionRedactedThinkingBlock(type="redacted_thinking", data=signature)
return ChatCompletionThinkingBlock(type="thinking", thinking=text, signature=signature)
def thinking_blocks_from_reasoning_items(
items: object,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
return tuple(
block for item in _object_sequence(items) if (block := thinking_block_from_reasoning_item(item)) is not None
)
def _block_encrypted_content(block: object) -> str | None:
mapping: Final = as_str_mapping(block)
if mapping is None:
return None
return encrypted_content_of_block(mapping)
def thinking_blocks_with_encrypted_reasoning(
thinking_blocks: object,
reasoning_items: object,
) -> tuple[Mapping[str, object], ...]:
"""Thinking blocks already on the message, plus any encrypted reasoning item they do not already carry."""
direct: Final = _iter_mappings(thinking_blocks)
represented: Final = frozenset(
content for block in direct if (content := _block_encrypted_content(block)) is not None
)
extra: Final = tuple(
block
for block in thinking_blocks_from_reasoning_items(_iter_mappings(reasoning_items))
if _block_encrypted_content(block) not in represented
)
return (*extra, *direct)
def is_encrypted_reasoning_block(block: object) -> bool:
"""A thinking or redacted_thinking block carrying Responses API encrypted reasoning.
@ -195,6 +282,14 @@ def anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
)
def _content_carries_encrypted_reasoning(content: object) -> bool:
return any(is_encrypted_reasoning_block(block) for block in _object_sequence(content))
def messages_carry_encrypted_reasoning(messages: Sequence[object]) -> bool:
return any(_content_carries_encrypted_reasoning(content) for content in anthropic_content_lists(messages))
def _strip_encrypted_reasoning_from_blocks(
content: object,
*,

1
litellm/litellm_core_utils/streaming_handler.py
View file

@ -1140,6 +1140,7 @@ class CustomStreamWrapper:
and model_response.choices[0].delta.role is not None
)
or (getattr(model_response.choices[0].delta, "reasoning_items", None) is not None)
or getattr(model_response.choices[0].delta, "thinking_blocks", None)
):
return True
else:

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

@ -1,5 +1,8 @@
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
import re
from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping, Sequence
from contextlib import AbstractContextManager, nullcontext
from dataclasses import dataclass
from functools import partial
from typing import (
TYPE_CHECKING,
Final,
@ -11,17 +14,25 @@ from pydantic import TypeAdapter
from typing_extensions import TypedDict
import litellm
from litellm._internal_context import resent_on
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.asyncify import run_async_function
from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
from litellm.litellm_core_utils.core_helpers import as_object_list, as_str_mapping, drops_unsupported_params
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
from litellm.litellm_core_utils.model_param_helper import peek_reasoning_summary_aliases
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
is_encrypted_reasoning_block,
thinking_blocks_with_encrypted_reasoning,
)
from litellm.litellm_core_utils.tool_fields import (
ROUTES_WITHOUT_MCP_SERVERS,
drops_mcp_servers,
tools_without_mcp_servers,
)
from litellm.llms.anthropic.common_utils import strip_encrypted_reasoning_blocks_from_anthropic_messages
from litellm.llms.anthropic.messages.adapters.transformation import (
AnthropicAdapter,
openai_thinking_routes_to_responses,
)
from litellm.llms.anthropic.messages.context_management import (
AnthropicContextManagementError,
@ -50,9 +61,9 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
)
from litellm.types.llms.openai import OpenAIWebSearchOptions
from litellm.types.utils import ModelResponse
from litellm.utils import get_model_info
if TYPE_CHECKING:
from litellm.chat_completions.dispatch import ChatResult
from litellm.router import Router
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
@ -70,22 +81,157 @@ class _CompletionKwargs(TypedDict, total=False, extra_items=object):
custom_llm_provider: str
_ANTHROPIC_MESSAGES: Final = TypeAdapter(list[dict[str, object]])
_UNVERIFIED_ENCRYPTED_CONTENT: Final = re.compile(r"encrypted content \S+ could not be verified", re.IGNORECASE)
_PrepareReplay: TypeAlias = "Callable[[_AnthropicMessages], _CompletionKwargs]"
@dataclass(frozen=True, slots=True)
class _ChatTargetReasoning:
effort: str
summary: str | None = None
def _provider_error_code(body: object) -> str | None:
mapping: Final = as_str_mapping(body)
if mapping is None:
return None
nested: Final = as_str_mapping(mapping.get("error"))
source: Final = nested if nested is not None else mapping
code: Final = source.get("code")
return code if isinstance(code, str) else None
def _logged_rejection_code(error: litellm.BadRequestError) -> str:
if error.code == "invalid_encrypted_content" or _provider_error_code(error.body) == "invalid_encrypted_content":
return "invalid_encrypted_content"
if "invalid_encrypted_content" in str(error):
return "invalid_encrypted_content"
return "could_not_be_verified"
def _completion_text(completion_kwargs: _CompletionKwargs, key: str) -> str:
value: Final = completion_kwargs.get(key)
return value if isinstance(value, str) else ""
def _message_carries_encrypted_reasoning(message: object) -> bool:
mapping: Final = as_str_mapping(message)
if mapping is None:
return False
blocks: Final = thinking_blocks_with_encrypted_reasoning(
mapping.get("thinking_blocks"), mapping.get("reasoning_items")
)
return any(is_encrypted_reasoning_block(block) for block in blocks)
def _sends_encrypted_reasoning(completion_kwargs: _CompletionKwargs) -> bool:
messages: Final = as_object_list(completion_kwargs.get("messages")) or ()
return any(_message_carries_encrypted_reasoning(message) for message in messages)
def _rejects_encrypted_reasoning(error: litellm.BadRequestError) -> bool:
error_message: Final = str(error)
return (
"invalid_encrypted_content" in error_message or _UNVERIFIED_ENCRYPTED_CONTENT.search(error_message) is not None
)
def _is_rejected_reasoning(error: Exception) -> bool:
return isinstance(error, litellm.BadRequestError) and _rejects_encrypted_reasoning(error)
def resent_without_rejected_reasoning(sent_encrypted_reasoning: bool) -> AbstractContextManager[None]:
return resent_on(_is_rejected_reasoning) if sent_encrypted_reasoning else nullcontext()
def messages_without_rejected_reasoning(
messages: Sequence[Mapping[str, object]],
error: litellm.BadRequestError,
*,
sent_encrypted_reasoning: bool,
provider: str,
model: str,
) -> "_AnthropicMessages | None":
if not sent_encrypted_reasoning or not _rejects_encrypted_reasoning(error):
return None
verbose_logger.warning(
"Backend rejected replayed encrypted reasoning, retrying once without it. "
"Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
"provider=%s model=%s status=%s code=%s",
provider,
model,
error.status_code,
_logged_rejection_code(error),
)
return _ANTHROPIC_MESSAGES.validate_python(
strip_encrypted_reasoning_blocks_from_anthropic_messages([dict(message) for message in messages])
)
def _messages_without_rejected_reasoning(
messages: _AnthropicMessages,
error: litellm.BadRequestError,
completion_kwargs: _CompletionKwargs,
sent_encrypted_reasoning: bool,
) -> "_AnthropicMessages | None":
return messages_without_rejected_reasoning(
messages,
error,
sent_encrypted_reasoning=sent_encrypted_reasoning,
provider=_completion_text(completion_kwargs, "custom_llm_provider"),
model=_completion_text(completion_kwargs, "model"),
)
async def _acompletion_without_rejected_reasoning(
completion_kwargs: _CompletionKwargs,
messages: _AnthropicMessages,
prepare_replay: _PrepareReplay,
) -> tuple[_CompletionKwargs, "ChatResult"]:
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return completion_kwargs, await litellm.acompletion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, completion_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = prepare_replay(replayed_messages)
return replay_kwargs, await litellm.acompletion(**replay_kwargs)
def _completion_without_rejected_reasoning(
completion_kwargs: _CompletionKwargs,
messages: _AnthropicMessages,
prepare_replay: _PrepareReplay,
) -> tuple[_CompletionKwargs, "ChatResult | Coroutine[object, object, ChatResult]"]:
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return completion_kwargs, litellm.completion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, completion_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = prepare_replay(replayed_messages)
return replay_kwargs, litellm.completion(**replay_kwargs)
def _is_compaction_block(block: object) -> bool:
mapping: Final = as_str_mapping(block)
return mapping is not None and mapping.get("type") == "compaction"
def _messages_have_compaction_block(messages: _AnthropicMessages) -> bool:
"""Return True when any message carries a ``compaction`` content block."""
for msg in messages:
content = msg.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "compaction":
return True
return False
return any(
any(map(_is_compaction_block, content)) for msg in messages if isinstance(content := msg.get("content"), list)
)
def _proxy_router_fallback() -> "Router | None":
@ -419,34 +565,17 @@ class LiteLLMMessagesToCompletionTransformationHandler:
If the user provides a `summary` field in the thinking dict, it is passed
through to the OpenAI reasoning params (opt-in per OpenAI spec).
"""
custom_llm_provider = completion_kwargs.get("custom_llm_provider")
if custom_llm_provider is None:
try:
_, inferred_provider, _, _ = litellm.utils.get_llm_provider(
model=cast(str, completion_kwargs.get("model"))
)
custom_llm_provider = inferred_provider
except Exception:
custom_llm_provider = None
if custom_llm_provider != "openai":
return
if not isinstance(thinking, dict) or thinking.get("type") != "enabled":
return
model: Final = completion_kwargs.get("model")
try:
model_info: Final = get_model_info(model=cast(str, model), custom_llm_provider=custom_llm_provider)
if model_info and model_info.get("supports_reasoning") is False:
# Model doesn't support reasoning/responses API, don't route
return
except Exception:
pass
custom_llm_provider: Final = completion_kwargs.get("custom_llm_provider")
if not isinstance(model, str) or thinking is None:
return
if not openai_thinking_routes_to_responses(
model, custom_llm_provider if isinstance(custom_llm_provider, str) else None, thinking
):
return
if isinstance(model, str) and model and "responses/" not in model:
local_model: Final = model.removeprefix(f"{custom_llm_provider}/")
completion_kwargs["model"] = f"{custom_llm_provider}/responses/{local_model}"
if "responses/" not in model:
completion_kwargs["model"] = f"openai/responses/{model.removeprefix('openai/')}"
auto_summary: Final = is_reasoning_auto_summary_enabled()
@ -677,6 +806,7 @@ class LiteLLMMessagesToCompletionTransformationHandler:
custom_llm_provider=target_provider,
api_base=api_base if isinstance(api_base, str) else None,
deployment_model_info=as_str_mapping(provider_kwargs.get("model_info")),
reasoning_summary=peek_reasoning_summary_aliases(provider_kwargs),
drop_params=drop_params,
)
@ -804,12 +934,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
parent_trace_id=sub_call_trace_id,
)
(
completion_kwargs,
tool_name_mapping,
) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
prepare: Final = partial(
LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
max_tokens=max_tokens,
messages=effective_messages,
model=model,
metadata=metadata,
stop_sequences=stop_sequences,
@ -824,8 +951,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
output_format=output_format,
extra_kwargs=kwargs,
)
completion_response: Final = await litellm.acompletion(**completion_kwargs)
initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
completion_kwargs, completion_response = await _acompletion_without_rejected_reasoning(
initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
)
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
if stream:
@ -951,12 +1080,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
effective_messages: Final = polyfill_result.messages if polyfill_result is not None else messages
effective_system: Final = polyfill_result.system if polyfill_result is not None else system
(
completion_kwargs,
tool_name_mapping,
) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
prepare: Final = partial(
LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
max_tokens=max_tokens,
messages=effective_messages,
model=model,
metadata=metadata,
stop_sequences=stop_sequences,
@ -971,8 +1097,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
output_format=output_format,
extra_kwargs=kwargs,
)
completion_response: Final = litellm.completion(**completion_kwargs)
initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
completion_kwargs, completion_response = _completion_without_rejected_reasoning(
initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
)
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
if stream:

27
litellm/llms/anthropic/messages/adapters/streaming_iterator.py
View file

@ -132,14 +132,21 @@ def _delta_payload_field(delta_type: StreamingContentBlockDeltaType) -> str:
assert_never(delta_type)
def _first_thinking_block_index(chunk: "ModelResponseStream") -> int | None:
def _first_thinking_block(chunk: "ModelResponseStream") -> Mapping[str, object] | None:
delta: Final = _optional_attr(chunk.choices[0], "delta") if chunk.choices else None
blocks: Final = _optional_attr_sequence(delta, "thinking_blocks")
first_block: Final = as_str_mapping(blocks[0]) if blocks else None
index: Final = first_block.get("index") if first_block is not None else None
return as_str_mapping(blocks[0]) if blocks else None
def _thinking_block_index(block: Mapping[str, object] | None) -> int | None:
index: Final = block.get("index") if block is not None else None
return index if isinstance(index, int) else None
def _closes_thinking_block(block: Mapping[str, object] | None) -> bool:
return block is not None and _thinking_block_index(block) is not None and bool(block.get("signature"))
def _optional_str_attr(obj: object, name: str) -> str | None:
value: Final = _optional_attr(obj, name)
return value if isinstance(value, str) else None
@ -632,6 +639,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
sent_content_block_finish: bool = False
current_content_block_type: Literal["text", "tool_use", "thinking", "redacted_thinking"] = "text"
current_thinking_block_index: int | None = None
current_thinking_block_closed: bool = False
sent_last_message: bool = False
holding_chunk: ContentBlockDelta | None = None
holding_stop_reason_chunk: MessageBlockDelta | None = None
@ -1575,9 +1583,16 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
original_name: Final = self.tool_name_mapping.get(truncated_name, truncated_name)
tool_block["name"] = original_name
thinking_block_index: Final = _first_thinking_block_index(chunk)
thinking_block: Final = _first_thinking_block(chunk)
thinking_block_index: Final = _thinking_block_index(thinking_block)
previous_thinking_block_index: Final = self.current_thinking_block_index
if thinking_block_index is not None or block_type != self.current_content_block_type:
follows_closed_thinking_block: Final = self.current_thinking_block_closed and block_type == "thinking"
self.current_thinking_block_closed = _closes_thinking_block(thinking_block)
if (
thinking_block_index is not None
or block_type != self.current_content_block_type
or follows_closed_thinking_block
):
self.current_thinking_block_index = thinking_block_index
if block_type != self.current_content_block_type:
@ -1585,7 +1600,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self.current_content_block_start = content_block_start
return True
if (
if follows_closed_thinking_block or (
thinking_block_index is not None
and previous_thinking_block_index is not None
and thinking_block_index != previous_thinking_block_index

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

@ -2,13 +2,14 @@ import copy
import hashlib
import json
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items, 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 (
@ -153,6 +154,7 @@ from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils i
)
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import (
reasoning_content_from_thinking_blocks,
thinking_blocks_with_encrypted_reasoning,
)
from litellm.litellm_core_utils.prompt_templates.tool_call_json_repair import (
parse_tool_call_arguments,
@ -258,27 +260,72 @@ def target_supports_mid_conversation_system(model: str | None, custom_llm_provid
return supports_mid_conversation_system(model=model, custom_llm_provider=custom_llm_provider)
def _is_azure_responses_target(
@dataclass(frozen=True, slots=True)
class ChatRouteInputs:
tools: Sequence[Mapping[str, object]] | None = None
reasoning_effort: str | Mapping[str, object] | None = None
api_base: str | None = None
thinking: Mapping[str, object] | None = None
reasoning_summary: object | None = None
_NO_ROUTE_INPUTS: Final = ChatRouteInputs()
def _inferred_provider(model: str) -> str | None:
try:
return litellm.utils.get_llm_provider(model=model)[1]
except Exception:
return None
def openai_thinking_routes_to_responses(
model: str | None,
custom_llm_provider: str | None,
thinking: Mapping[str, object] | None,
) -> bool:
if not model or thinking is None or thinking.get("type") != "enabled":
return False
provider: Final = custom_llm_provider if custom_llm_provider is not None else _inferred_provider(model)
if provider != "openai":
return False
try:
model_info: Final = litellm.get_model_info(model=model, custom_llm_provider=provider)
except Exception:
return True
return model_info.get("supports_reasoning") is not False
def _keeps_encrypted_reasoning_for_responses(
model: str | None,
custom_llm_provider: str | None,
deployment_model_info: Mapping[str, object] | None,
route: ChatRouteInputs,
) -> bool:
if not model:
return False
if custom_llm_provider not in (None, LlmProviders.AZURE.value):
return False
if custom_llm_provider is None and not model.startswith("azure/"):
return False
from litellm.responses.bridge_check import chat_completion_ends_in_responses_api
from litellm.responses.bridge_check import responses_api_bridge_check
resolved_model, provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)
model_info, _ = responses_api_bridge_check(
model=resolved_model,
custom_llm_provider=provider,
deployment_model_info=deployment_model_info,
return openai_thinking_routes_to_responses(
model, custom_llm_provider, route.thinking
) or chat_completion_ends_in_responses_api(
model,
custom_llm_provider,
deployment_model_info,
tools=route.tools,
reasoning_effort=route.reasoning_effort,
reasoning_summary=route.reasoning_summary,
api_base=route.api_base,
)
return model_info.get("mode") == "responses"
def _anthropic_thinking_blocks(source: object) -> tuple[Mapping[str, object], ...]:
return thinking_blocks_with_encrypted_reasoning(
getattr(source, "thinking_blocks", None),
getattr(source, "reasoning_items", None),
)
def _streamed_thinking_blocks(delta: object) -> tuple[Mapping[str, object], ...]:
return tuple(block for item in attribute_items(delta, "thinking_blocks") if (block := as_str_mapping(item)))
def _chat_tool_param(function_chunk: ChatCompletionToolParamFunctionChunk, tool: object) -> ChatCompletionToolParam:
@ -317,6 +364,7 @@ class AnthropicAdapter:
custom_llm_provider: str | None = None,
api_base: str | None = None,
deployment_model_info: Mapping[str, object] | None = None,
reasoning_summary: object | None = None,
drop_params: bool = True,
) -> tuple[ChatCompletionRequest | None, dict[str, str]]:
"""
@ -353,6 +401,7 @@ class AnthropicAdapter:
custom_llm_provider=custom_llm_provider,
api_base=api_base,
deployment_model_info=deployment_model_info,
reasoning_summary=reasoning_summary,
drop_params=drop_params,
report_reasoning_drops=True,
)
@ -551,11 +600,12 @@ class LiteLLMAnthropicMessagesAdapter:
custom_llm_provider: str | None = None,
deployment_model_info: Mapping[str, object] | None = None,
preserve_midturn_system: bool = False,
route: ChatRouteInputs = _NO_ROUTE_INPUTS,
) -> list[AllMessageValues]:
new_messages: Final[list[AllMessageValues]] = []
replayable_messages: Final = (
messages
if _is_azure_responses_target(model, custom_llm_provider, deployment_model_info)
if _keeps_encrypted_reasoning_for_responses(model, custom_llm_provider, deployment_model_info, route)
else strip_encrypted_reasoning_blocks_from_anthropic_messages(messages)
)
leading_count: Final = next(
@ -1338,6 +1388,7 @@ class LiteLLMAnthropicMessagesAdapter:
custom_llm_provider: str | None = None,
api_base: str | None = None,
deployment_model_info: Mapping[str, object] | None = None,
reasoning_summary: object | None = None,
preserve_midturn_system: bool = False,
drop_params: bool = True,
report_reasoning_drops: bool = False,
@ -1351,27 +1402,15 @@ class LiteLLMAnthropicMessagesAdapter:
for tools that exceeded OpenAI's 64-char limit
"""
# Debug: Processing Anthropic message request
new_messages: list[AllMessageValues] = []
tool_name_mapping: dict[str, str] = {}
## CONVERT ANTHROPIC MESSAGES TO OPENAI
messages_list: Final[list[AllAnthropicPassThroughMessageValues]] = cast(
list[AllAnthropicPassThroughMessageValues],
anthropic_message_request["messages"],
)
new_messages = self.translate_anthropic_messages_to_openai(
messages=messages_list,
model=anthropic_message_request.get("model"),
custom_llm_provider=custom_llm_provider,
deployment_model_info=deployment_model_info,
preserve_midturn_system=preserve_midturn_system,
)
## ADD SYSTEM MESSAGE TO MESSAGES
self._add_system_message_to_messages(new_messages, anthropic_message_request)
new_kwargs: Final[ChatCompletionRequest] = {
"model": anthropic_message_request["model"],
"messages": new_messages,
"messages": [],
}
## CONVERT METADATA (user_id + litellm metadata)
self._translate_metadata_to_openai(
@ -1398,6 +1437,24 @@ class LiteLLMAnthropicMessagesAdapter:
drop_params=drop_params,
report_reasoning_drops=report_reasoning_drops,
)
## CONVERT ANTHROPIC MESSAGES TO OPENAI, after tools and thinking pick the route
new_messages: Final = self.translate_anthropic_messages_to_openai(
messages=messages_list,
model=anthropic_message_request.get("model"),
custom_llm_provider=custom_llm_provider,
deployment_model_info=deployment_model_info,
preserve_midturn_system=preserve_midturn_system,
route=ChatRouteInputs(
tools=new_kwargs.get("tools"),
reasoning_effort=new_kwargs.get("reasoning_effort"),
api_base=api_base,
thinking=anthropic_message_request.get("thinking"),
reasoning_summary=reasoning_summary,
),
)
## ADD SYSTEM MESSAGE TO MESSAGES
self._add_system_message_to_messages(new_messages, anthropic_message_request)
new_kwargs["messages"] = new_messages
## CONVERT STOP_SEQUENCES
self._translate_stop_sequences_to_openai(
anthropic_message_request=anthropic_message_request,
@ -1502,9 +1559,9 @@ class LiteLLMAnthropicMessagesAdapter:
for choice in choices
):
new_content.extend(dict(_COMPACTION_BLOCK.validate_python(block)) for block in compaction_blocks)
# Handle thinking blocks first
if hasattr(choice.message, "thinking_blocks") and choice.message.thinking_blocks:
for thinking_block in choice.message.thinking_blocks:
signed_thinking_blocks = _anthropic_thinking_blocks(choice.message)
if signed_thinking_blocks:
for thinking_block in signed_thinking_blocks:
if thinking_block.get("type") == "thinking":
if is_empty_unsigned_thinking_block(thinking_block):
continue
@ -1816,24 +1873,24 @@ class LiteLLMAnthropicMessagesAdapter:
choice.delta
) is not None:
return "text", TextBlock(type="text", text="")
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
thinking_blocks = choice.delta.thinking_blocks or []
if len(thinking_blocks) > 0:
thinking_block = thinking_blocks[0]
if thinking_block["type"] == "redacted_thinking":
return "redacted_thinking", ChatCompletionRedactedThinkingBlock(
type="redacted_thinking", data=thinking_block.get("data") or ""
)
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
thinking_block = thinking_blocks[0]
if thinking_block["type"] == "redacted_thinking":
redacted_data = thinking_block.get("data")
return "redacted_thinking", ChatCompletionRedactedThinkingBlock(
type="redacted_thinking",
data=redacted_data if isinstance(redacted_data, str) else "",
)
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
assert isinstance(thinking, str)
assert isinstance(signature, str)
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking=thinking if signature else "", signature=signature
)
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking=thinking if signature else "", signature=signature
)
# OpenAI-compatible reasoning backends (e.g. vLLM/SGLang reasoning
# parsers) populate ``reasoning_content`` without ``thinking_blocks``.
# ``Delta`` deletes the ``thinking_blocks`` attribute when unset, so the
@ -1866,19 +1923,17 @@ class LiteLLMAnthropicMessagesAdapter:
for tool in choice.delta.tool_calls:
if tool.function is not None and tool.function.arguments is not None:
partial_json = (partial_json or "") + tool.function.arguments
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
thinking_blocks = choice.delta.thinking_blocks or []
if len(thinking_blocks) > 0:
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
assert isinstance(thinking, str)
assert isinstance(signature, str)
reasoning_content += thinking
reasoning_signature += signature
reasoning_content += thinking
reasoning_signature += signature
# Handle reasoning_content when thinking_blocks is not present
# This handles providers like OpenRouter that return reasoning_content
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "reasoning_content"):

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

@ -19,13 +19,14 @@ from litellm.integrations.anthropic_cache_control_hook import (
from litellm.litellm_core_utils.asyncify import (
run_async_function, # pyright: ignore[reportUnknownVariableType] # untyped helper
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
from litellm.litellm_core_utils.core_helpers import as_object_list, as_str_mapping, drops_unsupported_params
from litellm.litellm_core_utils.prompt_templates.prompt_cache_breakpoint_utils import (
cache_control_as_prompt_cache_breakpoint,
prompt_cache_breakpoint_with_readable_mode,
with_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.tool_fields import drops_mcp_servers, tools_without_mcp_servers
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
from litellm.responses.utils import auto_reasoning_summary
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic import (
@ -43,7 +44,12 @@ from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import CountContextManagement
from litellm.utils import ProviderConfigManager
from ..adapters.handler import context_management_explicitly_dropped, request_safeguards_evaluator
from ..adapters.handler import (
context_management_explicitly_dropped,
messages_without_rejected_reasoning,
request_safeguards_evaluator,
resent_without_rejected_reasoning,
)
from ..context_management import AnthropicContextManagementError, PolyfillResult, apply_context_management
from ..context_management.constants import COMPACT_EDIT_TYPE
from ..context_management.dispatcher import normalize_spec
@ -391,6 +397,76 @@ def _build_responses_kwargs(
return responses_kwargs
_ResponsesKwargsFor: TypeAlias = Callable[[AnthropicRequestMessages], Mapping[str, object]]
_ResponsesResult: TypeAlias = ResponsesAPIResponse | BaseResponsesAPIStreamingIterator
_SyncResponsesResult: TypeAlias = _ResponsesResult | Coroutine[object, object, _ResponsesResult]
def _is_encrypted_reasoning_item(item: object) -> bool:
mapping: Final = as_str_mapping(item)
return mapping is not None and mapping.get("type") == "reasoning" and bool(mapping.get("encrypted_content"))
def _sends_encrypted_reasoning(responses_kwargs: Mapping[str, object]) -> bool:
return any(map(_is_encrypted_reasoning_item, as_object_list(responses_kwargs.get("input")) or ()))
def _kwarg_text(responses_kwargs: Mapping[str, object], key: str) -> str:
value: Final = responses_kwargs.get(key)
return value if isinstance(value, str) else ""
def _messages_without_rejected_reasoning(
messages: AnthropicRequestMessages,
error: litellm.BadRequestError,
responses_kwargs: Mapping[str, object],
sent_encrypted_reasoning: bool,
) -> AnthropicRequestMessages | None:
return messages_without_rejected_reasoning(
messages,
error,
sent_encrypted_reasoning=sent_encrypted_reasoning,
provider=_kwarg_text(responses_kwargs, "custom_llm_provider"),
model=_kwarg_text(responses_kwargs, "model"),
)
async def _aresponses_without_rejected_reasoning(
messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
) -> tuple[Mapping[str, object], _ResponsesResult]:
responses_kwargs: Final = responses_kwargs_for(messages)
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return responses_kwargs, await litellm.aresponses(**responses_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, responses_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = responses_kwargs_for(replayed_messages)
return replay_kwargs, await litellm.aresponses(**replay_kwargs)
def _responses_without_rejected_reasoning(
messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
) -> tuple[Mapping[str, object], _SyncResponsesResult]:
responses_kwargs: Final = responses_kwargs_for(messages)
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return responses_kwargs, litellm.responses(**responses_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, responses_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = responses_kwargs_for(replayed_messages)
return replay_kwargs, litellm.responses(**replay_kwargs)
class LiteLLMMessagesToResponsesAPIHandler:
"""
Handles Anthropic /v1/messages requests for OpenAI / Azure models by
@ -422,28 +498,32 @@ class LiteLLMMessagesToResponsesAPIHandler:
polyfill_result: Final = await _polyfilled(
model=model, messages=messages, tools=tools, system=system, edits=polyfilled_edits, kwargs=kwargs
)
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=max_tokens,
messages=polyfill_result.messages if polyfill_result is not None else messages,
model=model,
context_management=native_context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
return _build_responses_kwargs(
max_tokens=max_tokens,
messages=sent_messages,
model=model,
context_management=native_context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
result: Final = await litellm.aresponses(**responses_kwargs)
responses_kwargs, result = await _aresponses_without_rejected_reasoning(
polyfill_result.messages if polyfill_result is not None else messages, responses_kwargs_for
)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(
@ -526,27 +606,31 @@ class LiteLLMMessagesToResponsesAPIHandler:
else None
)
polyfill_result: Final = ran if isinstance(ran, PolyfillResult) else None
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=max_tokens,
messages=polyfill_result.messages if polyfill_result is not None else messages,
model=model,
context_management=native_context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
result: Final = litellm.responses(**responses_kwargs)
def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
return _build_responses_kwargs(
max_tokens=max_tokens,
messages=sent_messages,
model=model,
context_management=native_context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
responses_kwargs, result = _responses_without_rejected_reasoning(
polyfill_result.messages if polyfill_result is not None else messages, responses_kwargs_for
)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(

33
litellm/main.py
View file

@ -28,6 +28,7 @@ from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, Union, cast, get_args
from litellm._internal_context import is_resent_failure
from litellm._logging import redact_string
from litellm._uuid import uuid
@ -6634,6 +6635,10 @@ def completion(
)
def _is_not_resent_failure(error: BaseException) -> bool:
return isinstance(error, Exception) and not is_resent_failure(error)
def completion_with_retries(*args, **kwargs):
"""
Executes a litellm.completion() with 3 retries
@ -6653,10 +6658,15 @@ def completion_with_retries(*args, **kwargs):
retryer = tenacity.Retrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.Retrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return retryer(original_function, *args, **kwargs)
@ -6679,10 +6689,15 @@ async def acompletion_with_retries(*args, **kwargs):
retryer = tenacity.AsyncRetrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.AsyncRetrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return await retryer(original_function, *args, **kwargs)
@ -6707,10 +6722,15 @@ def responses_with_retries(*args, **kwargs):
retryer = tenacity.Retrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.Retrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return retryer(original_function, *args, **kwargs)
@ -6734,10 +6754,15 @@ async def aresponses_with_retries(*args, **kwargs):
retryer = tenacity.AsyncRetrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.AsyncRetrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return await retryer(original_function, *args, **kwargs)

41
litellm/responses/bridge_check.py
View file

@ -4,7 +4,10 @@ from urllib.parse import urlsplit
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.get_llm_provider_logic import (
get_llm_provider,
resolve_model_and_provider_for_metadata,
)
from litellm.llms.azure_ai.chat.transformation import AzureAIGPT5Config
from litellm.llms.azure_ai.common_utils import (
azure_ai_supports_native_responses,
@ -291,3 +294,39 @@ def responses_api_bridge_check(
model = model.replace("responses/", "")
return model_info, model
def chat_completion_ends_in_responses_api(
model: str | None,
custom_llm_provider: str | None,
deployment_model_info: Mapping[str, object] | None = None,
*,
tools: Sequence[Mapping[str, object]] | None = None,
reasoning_effort: str | Mapping[str, object] | None = None,
reasoning_summary: object | None = None,
api_base: str | None = None,
) -> bool:
"""Whether chat completion for this model is served by a Responses API.
Encrypted reasoning can be replayed only there. A normal chat provider cannot
verify the block, so the messages adapter strips it. ``chatgpt``, a ``responses/``
model, a deployment pinned to ``mode: responses``, and a request whose tools or
reasoning effort the bridge sends to Responses keep it.
"""
if not model:
return False
try:
resolved_model, provider = resolve_model_and_provider_for_metadata(model, custom_llm_provider)
except Exception as error: # noqa: BLE001 # unknown model names must not fail message translation
verbose_logger.debug("responses replay check failed for %s: %s", model, error)
return False
model_info, _resolved = responses_api_bridge_check(
model=resolved_model,
custom_llm_provider=provider,
tools=tools,
reasoning_effort=reasoning_effort,
reasoning_summary=reasoning_summary,
api_base=api_base,
deployment_model_info=deployment_model_info,
)
return model_info.get("mode") == "responses"

34
litellm/utils.py
View file

@ -53,7 +53,7 @@ import litellm.litellm_core_utils
# audio_utils.utils is lazy-loaded - only imported when needed for transcription calls
import litellm.litellm_core_utils.json_validation_rule
from litellm._internal_context import is_internal_call
from litellm._internal_context import is_internal_call, is_resent_failure
from litellm._lazy_imports import (
_get_modified_max_tokens,
get_default_encoding,
@ -2303,12 +2303,16 @@ def client(original_function):
retry_policy=kwargs.get("retry_policy"),
)
kwargs["retry_policy"] = reset_retry_policy() # prevent infinite loops
litellm.num_retries = None # set retries to None to prevent infinite loops
if not is_resent_failure(e):
litellm.num_retries = None # set retries to None to prevent infinite loops
context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {})
is_completion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
if (
num_retries and not is_completion_litellm_router_call and not is_agentic_followup_failure(e)
num_retries
and not is_completion_litellm_router_call
and not is_agentic_followup_failure(e)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
if (
isinstance(e, openai.APIError)
@ -2338,11 +2342,15 @@ def client(original_function):
retry_policy=kwargs.get("retry_policy"),
)
kwargs["retry_policy"] = reset_retry_policy() # prevent infinite loops
litellm.num_retries = None # set retries to None to prevent infinite loops
if not is_resent_failure(e):
litellm.num_retries = None # set retries to None to prevent infinite loops
is_responses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
if (
num_retries and not is_responses_litellm_router_call and not is_agentic_followup_failure(e)
num_retries
and not is_responses_litellm_router_call
and not is_agentic_followup_failure(e)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
if (
isinstance(e, openai.APIError)
@ -2355,7 +2363,7 @@ def client(original_function):
end_time = datetime.datetime.now()
# LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
if logging_obj:
if logging_obj and not is_resent_failure(e):
logging_obj.failure_handler(
e, traceback_exception, start_time, end_time
) # DO NOT MAKE THREADED - router retry fallback relies on this!
@ -2620,7 +2628,7 @@ def client(original_function):
# Reuse the timestamp taken right when the deployment call itself failed, before
# the failure hook ran, so a slow callback doesn't inflate the reported duration.
end_time = _deployment_call_end_time if _deployment_call_end_time is not None else datetime.datetime.now() # noqa: DTZ005 # matches the naive datetimes this whole function already times start_time/end_time with
if logging_obj and not _is_litellm_internal_call:
if logging_obj and not _is_litellm_internal_call and not is_resent_failure(e):
unbind_budget_reservation_from_callbacks(logging_obj.litellm_params)
try:
logging_obj.failure_handler(
@ -2645,6 +2653,7 @@ def client(original_function):
and not is_acompletion_litellm_router_call
and not isinstance(e, ImportError)
and not is_agentic_followup_failure(e)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
try:
litellm.num_retries = None # set retries to None to prevent infinite loops
@ -2655,8 +2664,9 @@ def client(original_function):
elif isinstance(e, openai.APIError): # generic api error
kwargs["retry_strategy"] = "constant_retry"
result = await litellm.acompletion_with_retries(*args, **kwargs)
except Exception:
pass
except Exception as retry_error:
if is_resent_failure(retry_error):
raise
else:
return result
elif (
@ -2679,6 +2689,7 @@ def client(original_function):
and not is_aresponses_litellm_router_call
and not isinstance(e, ImportError)
and not is_agentic_followup_failure(e)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
try:
litellm.num_retries = None # set retries to None to prevent infinite loops
@ -2689,8 +2700,9 @@ def client(original_function):
elif isinstance(e, openai.APIError): # generic api error
kwargs["retry_strategy"] = "constant_retry"
result = await litellm.aresponses_with_retries(*args, **kwargs)
except Exception:
pass
except Exception as retry_error:
if is_resent_failure(retry_error):
raise
else:
return result

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

@ -3837,6 +3837,386 @@ def test_reasoning_items_streaming_emitted_on_response_completed():
assert ri["summary"][0]["text"] == summary_text
def test_reasoning_output_item_done_emits_signed_thinking_blocks_the_stream_keeps():
"""A chatgpt /v1/messages stream drops the reasoning item unless this delta carries thinking_blocks.
response.output_item.done is the only event that has encrypted_content. The chat delta has to
expose it as a signed thinking block, and that block alone has to count as stream content.
"""
from unittest.mock import MagicMock
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.llms.anthropic.messages.adapters.transformation import LiteLLMAnthropicMessagesAdapter
encrypted: Final = "gAAAA_round_trip"
done: Final = {
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_done",
"encrypted_content": encrypted,
"summary": [],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
blocks: Final = result.choices[0].delta.thinking_blocks
assert blocks is not None
assert blocks[0]["type"] == "redacted_thinking"
assert blocks[0]["data"] == encrypted_reasoning_signature(encrypted)
assert blocks[0]["index"] == 0
assert getattr(result.choices[0].delta, "reasoning_content", None) in (None, "")
wrapper: Final = CustomStreamWrapper(
completion_stream=None,
model="chatgpt/gpt-6.1-sol",
logging_obj=MagicMock(),
custom_llm_provider="chatgpt",
)
assert wrapper.is_chunk_non_empty({}, result, {}) is True
block_type, start = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_chunk_to_anthropic_content_block(
choices=result.choices
)
assert block_type == "redacted_thinking"
assert start["data"] == encrypted_reasoning_signature(encrypted)
_SIGNED_REASONING_STREAM_BODY: Final[Mapping[str, object]] = {
**_searched_responses_body("azure/gpt-6-sol"),
"output": [
{"type": "reasoning", "id": "rs_signed", "summary": [], "encrypted_content": "gAAAA_consumer"},
_PONG_MESSAGE,
],
}
def _signed_reasoning_sse() -> str:
events: Final = (
{
"type": "response.created",
"response": {**_SIGNED_REASONING_STREAM_BODY, "status": "in_progress", "output": [], "usage": None},
},
{"type": "response.output_item.done", "output_index": 0, "item": _SIGNED_REASONING_STREAM_BODY["output"][0]},
{
"type": "response.output_text.delta",
"item_id": "msg_logged_call_type",
"output_index": 1,
"content_index": 0,
"delta": "pong",
},
{"type": "response.completed", "response": _SIGNED_REASONING_STREAM_BODY},
)
return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events)
_AZURE_BRIDGE_KWARGS: Final[Mapping[str, object]] = {
"api_base": "https://example.invalid",
"api_version": "2025-04-01-preview",
"api_key": "test-key",
}
@pytest.mark.asyncio
async def test_streamed_reasoning_signature_reaches_messages_clients_and_never_chat_completions_deltas(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
with respx.mock(assert_all_called=True) as router:
router.post(url__startswith="https://example.invalid/openai/responses").mock(
side_effect=lambda _request: httpx.Response(
200, text=_signed_reasoning_sse(), headers={"content-type": "text/event-stream"}
)
)
chat_stream: Final = await litellm.acompletion(
model="azure/gpt-6-sol",
messages=[{"role": "user", "content": "ping"}],
tools=[_GET_WEATHER_TOOL],
stream=True,
**_AZURE_BRIDGE_KWARGS,
)
chat_deltas: Final = [chunk.choices[0].delta async for chunk in chat_stream if chunk.choices]
messages_stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-6-sol",
messages=[{"role": "user", "content": "ping"}],
tools=[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
max_tokens=64,
stream=True,
**_AZURE_BRIDGE_KWARGS,
)
sse: Final = b"".join([chunk async for chunk in messages_stream]).decode()
assert [getattr(delta, "thinking_blocks", None) for delta in chat_deltas if getattr(delta, "thinking_blocks", None)] == []
assert "".join(delta.content or "" for delta in chat_deltas) == "pong"
messages_events: Final = [json.loads(line[len("data: ") :]) for line in sse.splitlines() if line.startswith("data: ")]
started: Final = [event["content_block"] for event in messages_events if event.get("type") == "content_block_start"]
assert [block["type"] for block in started] == ["redacted_thinking", "text"]
assert started[0]["data"] == encrypted_reasoning_signature("gAAAA_consumer")
def test_reasoning_output_item_done_with_summary_signs_the_thinking_text():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
encrypted: Final = "gAAAA_with_summary"
done: Final = {
"type": "response.output_item.done",
"output_index": 2,
"item": {
"type": "reasoning",
"id": "rs_sum",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
block: Final = result.choices[0].delta.thinking_blocks[0]
assert block["type"] == "thinking"
assert block["thinking"] == "Oslo is colder."
assert block["signature"] == encrypted_reasoning_signature(encrypted)
assert block["index"] == 2
def test_streamed_reasoning_summary_reaches_messages_clients_once_with_its_signature():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.messages.adapters.streaming_iterator import AnthropicStreamWrapper
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
encrypted: Final = "gAAAA_streamed_summary"
events: Final = (
{"type": "response.reasoning_summary_part.added", "output_index": 0, "summary_index": 0},
{"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "Oslo is "},
{"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "colder."},
{
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_streamed",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
},
},
{"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
)
parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
)
chunks: Final = (
*(parser.chunk_parser(event) for event in events),
ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
)
sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="chatgpt/gpt-5.5"))
deltas: Final = [event["delta"] for event in sse_events if event.get("type") == "content_block_delta"]
assert "".join(d.get("thinking", "") for d in deltas if d["type"] == "thinking_delta") == "Oslo is colder."
assert [d["signature"] for d in deltas if d["type"] == "signature_delta"] == [
encrypted_reasoning_signature(encrypted)
]
def test_reasoning_item_repeated_at_stream_end_reaches_messages_clients_once():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.messages.adapters.streaming_iterator import AnthropicStreamWrapper
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
encrypted: Final = "gAAAA_completed_repeat"
reasoning_item: Final = {"type": "reasoning", "id": "rs_repeat", "summary": [], "encrypted_content": encrypted}
events: Final = (
{"type": "response.output_item.done", "output_index": 0, "item": reasoning_item},
{"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
)
parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
)
chunks: Final = (
*(parser.chunk_parser(event) for event in events),
ModelResponseStream(
choices=[StreamingChoices(index=0, delta=Delta(reasoning_items=[reasoning_item]), finish_reason=None)]
),
ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
)
sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="azure/gpt-5.4-nano"))
started: Final = [event["content_block"] for event in sse_events if event.get("type") == "content_block_start"]
assert [block["type"] for block in started] == ["redacted_thinking", "text"]
assert started[0]["data"] == encrypted_reasoning_signature(encrypted)
def test_reasoning_output_item_done_without_encrypted_content_stays_empty():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
done: Final = {
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_plain",
"summary": [{"type": "summary_text", "text": "already streamed"}],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
assert getattr(result.choices[0].delta, "thinking_blocks", None) in (None, [])
def test_transform_response_hands_encrypted_reasoning_to_messages_clients_only():
from unittest.mock import Mock
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.messages.adapters.transformation import (
LiteLLMAnthropicMessagesAdapter,
)
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import ModelResponse, Usage
encrypted: Final = "gAAAA_non_stream"
reasoning: Final = ResponseReasoningItem(
id="rs_tool",
summary=[],
type="reasoning",
content=None,
encrypted_content=encrypted,
status=None,
)
function_call: Final = ResponseFunctionToolCall(
id="fc_1",
type="function_call",
status="completed",
arguments='{"city":"Oslo"}',
call_id="call_1",
name="get_weather",
)
raw: Final = ResponsesAPIResponse.model_construct(
id="resp_tool",
created_at=1,
error=None,
incomplete_details=None,
model="gpt-6.1-sol",
output=[reasoning, function_call],
status="completed",
usage=None,
)
model_response: Final = ModelResponse(
id="chatcmpl-tool",
created=1,
model=None,
object="chat.completion",
system_fingerprint=None,
choices=[],
usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
)
result: Final = LiteLLMResponsesTransformationHandler().transform_response(
model="gpt-6.1-sol",
raw_response=raw,
model_response=model_response,
logging_obj=Mock(),
request_data={"model": "gpt-6.1-sol"},
messages=[{"role": "user", "content": "Which city?"}],
optional_params={},
litellm_params={},
encoding=Mock(),
)
message: Final = result.choices[0].message
assert message.reasoning_items[0]["encrypted_content"] == encrypted
assert "thinking_blocks" not in result.model_dump()["choices"][0]["message"]
assert message.tool_calls[0].function.name == "get_weather"
anthropic_response: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(
response=result, tool_name_mapping=None
)
content: Final = anthropic_response["content"]
assert [block["type"] for block in content] == ["redacted_thinking", "tool_use"]
assert content[0]["data"] == encrypted_reasoning_signature(encrypted)
@pytest.mark.parametrize(
("call_type", "kept_ids"),
[
pytest.param("anthropic_messages", ["rs_first", "rs_second"], id="messages_client_keeps_every_item"),
pytest.param("acompletion", ["rs_second"], id="chat_client_keeps_the_last_item"),
],
)
def test_non_streamed_tool_turn_hands_messages_clients_every_reasoning_item(
call_type: str, kept_ids: list[str]
) -> None:
from types import SimpleNamespace
from unittest.mock import Mock
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import ModelResponse, Usage
def reasoning(item_id: str) -> ResponseReasoningItem:
return ResponseReasoningItem(
id=item_id, summary=[], type="reasoning", content=None, encrypted_content=f"gAAAA_{item_id}", status=None
)
def call(index: int, city: str) -> ResponseFunctionToolCall:
return ResponseFunctionToolCall(
id=f"fc_{index}",
type="function_call",
status="completed",
arguments=json.dumps({"city": city}),
call_id=f"call_{index}",
name="get_weather",
)
raw: Final = ResponsesAPIResponse.model_construct(
id="resp_two_reasoning",
created_at=1,
error=None,
incomplete_details=None,
model="gpt-5.4-nano",
output=[reasoning("rs_first"), call(1, "Oslo"), reasoning("rs_second"), call(2, "Lyon")],
status="completed",
usage=None,
)
result: Final = LiteLLMResponsesTransformationHandler().transform_response(
model="gpt-5.4-nano",
raw_response=raw,
model_response=ModelResponse(
id="chatcmpl-two-reasoning",
created=1,
model=None,
object="chat.completion",
system_fingerprint=None,
choices=[],
usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
),
logging_obj=SimpleNamespace(call_type=call_type),
request_data={"model": "gpt-5.4-nano"},
messages=[{"role": "user", "content": "Oslo or Lyon?"}],
optional_params={},
litellm_params={"custom_llm_provider": "azure"},
encoding=Mock(),
)
message: Final = result.choices[0].message
assert [item["id"] for item in message.reasoning_items] == kept_ids
assert [tool_call.function.name for tool_call in message.tool_calls] == ["get_weather", "get_weather"]
def test_streaming_function_call_tool_id_for_degenerate_call_id():
"""In streaming, Bedrock Mantle's function_call event carries a unique ``id``
(``fc_...``) and a non-unique, index-based ``call_id`` (``call_0``). For that

31
tests/unit/litellm_core_utils/test_streaming_handler.py
View file

@ -276,6 +276,37 @@ def test_is_chunk_non_empty(initialized_custom_stream_wrapper: CustomStreamWrapp
)
def test_is_chunk_non_empty_keeps_thinking_blocks_without_other_content(
initialized_custom_stream_wrapper: CustomStreamWrapper,
):
"""Encrypted reasoning arrives as thinking_blocks and nothing else, so that delta has to be yielded."""
chunk: Final = ModelResponseStream(
id="chunk-thinking",
object="chat.completion.chunk",
created=1,
model="chatgpt/gpt-6.1-sol",
choices=[
StreamingChoices(
index=0,
finish_reason=None,
delta=Delta(
thinking_blocks=[
{"type": "redacted_thinking", "data": "litellm_encrypted_reasoning:gAAAA"}
]
),
)
],
)
assert (
initialized_custom_stream_wrapper.is_chunk_non_empty(
completion_obj={},
model_response=chunk,
response_obj={},
)
is True
)
def test_is_chunk_non_empty_with_annotations(
initialized_custom_stream_wrapper: CustomStreamWrapper,
):

717
tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py
View file

@ -2,7 +2,8 @@ import base64
import functools
import json
import logging
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from contextlib import nullcontext
from types import MappingProxyType
from typing import Any, Final, cast
@ -11,6 +12,7 @@ import pytest
import respx
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_PLACEHOLDER,
)
@ -34,6 +36,7 @@ from litellm.llms.anthropic.messages.native.mid_conversation_system import (
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.types.llms.anthropic import (
AnthopicMessagesAssistantMessageParam,
AnthropicMessagesRequest,
AnthropicMessagesUserMessageParam,
)
from litellm.types.llms.openai import ChatCompletionAssistantToolCall
@ -616,6 +619,218 @@ def test_translate_anthropic_messages_to_openai_drops_bridge_encrypted_reasoning
assert [b["signature"] for b in result[2]["thinking_blocks"]] == ["EqQBCkYIAxgCIkA_signed"]
def test_chatgpt_messages_keep_encrypted_reasoning_and_replay_it():
"""chatgpt /v1/messages ends in the Responses API, so the signed block has to survive into the next request."""
from litellm.completion_extras.litellm_responses_transformation.transformation import (
LiteLLMResponsesTransformationHandler,
)
summary_encrypted: Final = "gAAAA_summary"
bytes_encrypted: Final = "gAAAA_bytes"
messages: Final = [
AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
AnthopicMessagesAssistantMessageParam(
role="assistant",
content=[
{
"type": "thinking",
"thinking": "Oslo is colder.",
"signature": encrypted_reasoning_signature(summary_encrypted),
},
{"type": "redacted_thinking", "data": encrypted_reasoning_signature(bytes_encrypted)},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
),
AnthropicMessagesUserMessageParam(
role="user",
content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
),
]
kept: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="chatgpt/gpt-6.1-sol",
custom_llm_provider="chatgpt",
)
assistant: Final = kept[1]
assert [block["type"] for block in assistant["thinking_blocks"]] == ["thinking", "redacted_thinking"]
assert assistant["thinking_blocks"][0]["signature"] == encrypted_reasoning_signature(summary_encrypted)
assert assistant["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
responses_input, _instructions = LiteLLMResponsesTransformationHandler().convert_chat_completion_messages_to_responses_api(
kept
)
reasoning: Final = tuple(item for item in responses_input if item.get("type") == "reasoning")
assert [item["encrypted_content"] for item in reasoning] == [summary_encrypted, bytes_encrypted]
types: Final = [item.get("type") for item in responses_input]
assert types.index("reasoning") < types.index("function_call")
stripped: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="gpt-4o",
custom_llm_provider="openai",
)
assert not stripped[1].get("thinking_blocks")
prefixed: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="responses/gpt-5-mini",
custom_llm_provider="openai",
)
assert prefixed[1]["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
def test_translate_openai_reasoning_items_become_signed_thinking_blocks():
encrypted: Final = "gAAAA_outbound"
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content="Oslo.",
reasoning_items=[
{
"type": "reasoning",
"id": "rs_1",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
}
],
tool_calls=[
ChatCompletionAssistantToolCall(
id="call_1",
type="function",
function=Function(name="get_weather", arguments='{"city": "Oslo"}'),
)
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert result[0]["type"] == "thinking"
assert result[0]["thinking"] == "Oslo is colder."
assert result[0]["signature"] == encrypted_reasoning_signature(encrypted)
assert result[1]["type"] == "text"
assert result[2]["type"] == "tool_use"
def test_translate_openai_reasoning_items_without_summary_become_redacted_thinking():
encrypted: Final = "gAAAA_bytes_only"
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content=None,
reasoning_items=[
{"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert result == [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature(encrypted)},
]
def test_translate_openai_reasoning_items_are_not_duplicated_when_thinking_blocks_already_carry_them():
encrypted: Final = "gAAAA_once"
signature: Final = encrypted_reasoning_signature(encrypted)
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content="Oslo.",
thinking_blocks=[{"type": "redacted_thinking", "data": signature}],
reasoning_items=[
{"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert [block["type"] for block in result] == ["redacted_thinking", "text"]
def _replayed_reasoning_request(tools: list[dict[str, object]]) -> AnthropicMessagesRequest:
return AnthropicMessagesRequest(
model="azure/gpt-5.4-nano",
max_tokens=64,
messages=[
AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
AnthopicMessagesAssistantMessageParam(
role="assistant",
content=[
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAAA_azure_tools")},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
),
AnthropicMessagesUserMessageParam(
role="user",
content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
),
],
tools=tools,
)
@pytest.mark.parametrize(
("tools", "kept"),
[
pytest.param(
[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
True,
id="function_tools_bridge_to_responses",
),
pytest.param([], False, id="no_tools_stay_on_chat"),
],
)
def test_encrypted_reasoning_is_kept_when_the_request_tools_route_it_to_responses(
tools: list[dict[str, object]], kept: bool
) -> None:
openai_request, _mapping = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
_replayed_reasoning_request(tools),
custom_llm_provider="azure",
api_base="https://keep-check.openai.azure.com",
)
assistant: Final = openai_request["messages"][1]
assert bool(assistant.get("thinking_blocks")) is kept
@pytest.mark.parametrize(
("deployment_params", "kept"),
[
pytest.param({"reasoning_summary": "auto"}, True, id="deployment_summary_bridges_to_responses"),
pytest.param({"extra_body": {"reasoningSummary": "auto"}}, True, id="extra_body_summary_bridges_to_responses"),
pytest.param({}, False, id="no_summary_stays_on_chat"),
],
)
def test_encrypted_reasoning_is_kept_when_the_deployment_reasoning_summary_routes_it_to_responses(
deployment_params: dict[str, object], kept: bool
) -> None:
from litellm.llms.anthropic.messages.adapters.handler import (
LiteLLMMessagesToCompletionTransformationHandler,
)
request: Final = _replayed_reasoning_request([])
completion_kwargs, _mapping = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
max_tokens=64,
messages=cast(list[dict[str, object]], request["messages"]),
model="azure/gpt-5.1",
thinking={"type": "enabled", "budget_tokens": 2048},
extra_kwargs={"api_base": "https://keep-check.openai.azure.com", **deployment_params},
)
assistant: Final = cast(list[Mapping[str, object]], completion_kwargs["messages"])[1]
assert bool(assistant.get("thinking_blocks")) is kept
def test_translate_anthropic_messages_to_openai_sets_reasoning_content():
"""Reasoning-aware chat providers read reasoning_content, so thinking text must land there.
@ -5578,3 +5793,503 @@ async def test_chat_bridge_tool_result_text_block_without_text_reaches_the_model
],
}
]
_RETRY_API_BASE: Final = "https://encrypted-retry.openai.azure.com"
_RETRY_URL: Final = r".*encrypted-retry\.openai\.azure\.com/openai/.*responses.*"
_RETRY_TOOLS: Final = [{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}]
_RETRY_REJECTION: Final = httpx.Response(
400,
json={
"error": {
"message": "The encrypted content gAAA_foreign could not be verified.",
"type": "invalid_request_error",
"param": None,
"code": "invalid_encrypted_content",
}
},
)
_UNRELATED_REJECTION: Final = httpx.Response(
400, json={"error": {"message": "max_output_tokens is too small", "type": "invalid_request_error"}}
)
_UNRELATED_UNVERIFIED_REJECTION: Final = httpx.Response(
400, json={"error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}}
)
_RETRY_REPLY: Final = {
"id": "resp_retry",
"object": "response",
"created_at": 1700000000,
"status": "completed",
"model": "gpt-5.4-nano",
"output": [
{
"type": "message",
"id": "msg_retry",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
],
"usage": {"input_tokens": 10, "output_tokens": 2, "total_tokens": 12},
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
}
def _history_with_foreign_reasoning() -> list[dict[str, object]]:
return [
{"role": "user", "content": "Which city is colder?"},
{
"role": "assistant",
"content": [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}]},
]
def _sent_reasoning_items(request: httpx.Request) -> list[object]:
sent: Final = json.loads(request.read())
return [item for item in sent["input"] if item.get("type") == "reasoning"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter,
caplog: pytest.LogCaptureFixture,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
response: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
warnings: Final = tuple(
record.getMessage()
for record in caplog.records
if record.levelno >= logging.WARNING and record.name == "LiteLLM"
)
retry_warnings: Final = tuple(message for message in warnings if "retrying once without it" in message)
assert route.call_count == 2
assert len(_sent_reasoning_items(route.calls[0].request)) == 1
assert _sent_reasoning_items(route.calls[1].request) == []
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
assert retry_warnings == (
"Backend rejected replayed encrypted reasoning, retrying once without it. "
"Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
"provider=azure model=azure/gpt-5.4-nano status=400 code=invalid_encrypted_content",
)
assert all(
"gAAA_foreign" not in message and "litellm_encrypted_reasoning:" not in message for message in warnings
)
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert route.call_count == 2
assert _sent_reasoning_items(route.calls[1].request) == []
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_with_num_retries_send_the_rejected_reasoning_only_once(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_with_global_num_retries_send_the_rejected_reasoning_once_and_keep_the_setting(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "num_retries", 2)
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
assert litellm.num_retries == 2
_SERVER_ERROR: Final = httpx.Response(
500, json={"error": {"message": "The server had an error while processing your request.", "type": "server_error"}}
)
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
class _FailureLog(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in _UPSTREAM_REJECTION_PHRASES if phrase in error), error),
)
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
_UPSTREAM_REJECTION_PHRASES: Final = ("could not be verified", "max_output_tokens is too small")
_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
pytest.param(httpx.Response(200, json=_RETRY_REPLY), (), id="served"),
pytest.param(_UNRELATED_REJECTION, ("max_output_tokens is too small",), id="rejected"),
)
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
async def test_async_messages_log_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert failure_log.rejections == logged_rejections
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
def test_sync_messages_log_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert failure_log.rejections == logged_rejections
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(
("messages", "rejection"),
[
pytest.param(
[{"role": "user", "content": "Which city is colder?"}],
_RETRY_REJECTION,
id="no_encrypted_reasoning_to_drop",
),
pytest.param(_history_with_foreign_reasoning(), _UNRELATED_REJECTION, id="unrelated_rejection"),
pytest.param(
_history_with_foreign_reasoning(), _UNRELATED_UNVERIFIED_REJECTION, id="unrelated_unverified_rejection"
),
],
)
async def test_messages_do_not_retry_other_rejections(
respx_mock: respx.MockRouter, messages: list[dict[str, object]], rejection: httpx.Response
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(return_value=rejection)
with pytest.raises(litellm.BadRequestError):
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=messages,
)
assert route.call_count == 1
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_messages_do_not_retry_when_the_translated_request_already_dropped_the_reasoning(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_CHAT_BRIDGE_URL).mock(return_value=_RETRY_REJECTION)
with pytest.raises(litellm.BadRequestError):
await litellm.anthropic.messages.acreate(
model=_CHAT_BRIDGE_MODEL,
api_base=_CHAT_BRIDGE_API_BASE,
api_key="fake-key",
max_tokens=64,
messages=_history_with_foreign_reasoning(),
)
assert route.call_count == 1
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_openai_thinking_on_the_chat_adapter_replays_the_reasoning_its_responses_route_minted(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", True)
route: Final = respx_mock.post(url__regex=r".*openai-thinking\.example/v1/responses").mock(
return_value=httpx.Response(200, json=_RETRY_REPLY)
)
await litellm.anthropic.messages.acreate(
model="openai/gpt-5.4-nano",
api_base="https://openai-thinking.example/v1",
api_key="fake-key",
max_tokens=2048,
thinking={"type": "enabled", "budget_tokens": 1024},
messages=[
{"role": "user", "content": "Which city is colder?"},
{
"role": "assistant",
"content": [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_same_org")},
{"type": "text", "text": "Oslo."},
],
},
{"role": "user", "content": "And by how much?"},
],
)
assert route.call_count == 1
assert [item["encrypted_content"] for item in _sent_reasoning_items(route.calls[0].request)] == ["gAAA_same_org"]
def _responses_sse(events: Sequence[Mapping[str, object]]) -> bytes:
return b"".join(
f"event: {event['type']}\ndata: {json.dumps({**event, 'sequence_number': number})}\n\n".encode()
for number, event in enumerate(events)
)
def _reasoning_item(item_id: str, encrypted: str, summary: str = "") -> dict[str, object]:
return {
"type": "reasoning",
"id": item_id,
"summary": [{"type": "summary_text", "text": summary}] if summary else [],
"encrypted_content": encrypted,
}
def _streamed_blocks(frames: str) -> list[tuple[str, str, str | None]]:
events: Final = tuple(
json.loads(line.removeprefix("data: ")) for line in frames.splitlines() if line.startswith("data: ")
)
deltas: Final = tuple(event for event in events if event["type"] == "content_block_delta")
def block_deltas(index: int) -> tuple[Mapping[str, str], ...]:
return tuple(delta["delta"] for delta in deltas if delta["index"] == index)
return [
(
event["content_block"]["type"],
"".join(delta.get("thinking", "") for delta in block_deltas(event["index"])),
next(
(delta["signature"] for delta in block_deltas(event["index"]) if delta["type"] == "signature_delta"),
None,
),
)
for event in events
if event["type"] == "content_block_start"
]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_streamed_reasoning_summaries_stay_with_their_own_encrypted_item(
respx_mock: respx.MockRouter,
) -> None:
response: Final = {**_RETRY_REPLY, "id": "resp_two_summaries", "status": "in_progress", "output": []}
message: Final = {
"type": "message",
"id": "msg_two_summaries",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
first: Final = _reasoning_item("rs_1", "gAAA_first", "Oslo is further north.")
second: Final = _reasoning_item("rs_2", "gAAA_second", "Check the forecast.")
completed: Final = {**response, "status": "completed", "output": [first, second, message]}
def summary_delta(output_index: int, item_id: str, text: str) -> dict[str, object]:
return {
"type": "response.reasoning_summary_text.delta",
"output_index": output_index,
"item_id": item_id,
"summary_index": 0,
"delta": text,
}
events: Final = (
{"type": "response.created", "response": response},
{"type": "response.output_item.added", "output_index": 0, "item": _reasoning_item("rs_1", "")},
summary_delta(0, "rs_1", "Oslo is further north."),
{"type": "response.output_item.done", "output_index": 0, "item": first},
{"type": "response.output_item.added", "output_index": 1, "item": _reasoning_item("rs_2", "")},
summary_delta(1, "rs_2", "Check the forecast."),
{"type": "response.output_item.done", "output_index": 1, "item": second},
{"type": "response.output_item.added", "output_index": 2, "item": {**message, "content": []}},
{
"type": "response.output_text.delta",
"output_index": 2,
"item_id": "msg_two_summaries",
"content_index": 0,
"delta": "Oslo",
},
{"type": "response.output_item.done", "output_index": 2, "item": message},
{"type": "response.completed", "response": completed},
)
respx_mock.post(url__regex=_RETRY_URL).mock(
return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
)
stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
stream=True,
messages=[{"role": "user", "content": "Which city is colder?"}],
)
frames: Final = b"".join([chunk async for chunk in stream]).decode()
assert _streamed_blocks(frames) == [
("thinking", "Oslo is further north.", encrypted_reasoning_signature("gAAA_first")),
("thinking", "Check the forecast.", encrypted_reasoning_signature("gAAA_second")),
("text", "", None),
]

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

@ -3,20 +3,24 @@ import json
import os
import sys
from collections.abc import AsyncIterator, Mapping
from contextlib import nullcontext
from itertools import chain
from typing import Final, cast
from unittest.mock import AsyncMock, patch
import httpx
import pytest
import respx
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_BOUNDARY,
TOOL_RESULT_IMAGE_PLACEHOLDER,
)
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.messages.responses_adapters.handler import (
LiteLLMMessagesToResponsesAPIHandler,
_build_responses_kwargs,
@ -1135,3 +1139,291 @@ 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
REJECTED_REPLAYED_REASONING: Final = {
"error": {
"message": "The encrypted content gAAA_foreign could not be verified.",
"type": "invalid_request_error",
"param": None,
"code": "invalid_encrypted_content",
}
}
UNVERIFIED_ORGANIZATION: Final = {
"error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}
}
def _history(*, replayed_reasoning: bool) -> list[dict[str, object]]:
reasoning: Final = (
({"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},)
if replayed_reasoning
else ()
)
return [
{"role": "user", "content": "List the logs."},
{"role": "assistant", "content": [*reasoning, {"type": "tool_use", "id": "call_1", "name": "Bash", "input": {}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "app.log"}]},
]
def _replayed_reasoning_counts(upstream: respx.Route) -> list[int]:
return [
sum(item.get("type") == "reasoning" for item in json.loads(call.request.content)["input"])
for call in upstream.calls
]
def _upstream(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, first: httpx.Response, stream: bool):
return _upstream_after(respx_mock, monkeypatch, (first,), stream)
def _upstream_after(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, failures: tuple[httpx.Response, ...], stream: bool
) -> respx.Route:
monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-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=[*failures, served])
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True])
async def test_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, stream: bool
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), stream)
response: Final = await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
stream=stream,
custom_llm_provider="openai",
)
events: Final = (
[json.loads(chunk.decode().split("data: ", 1)[1]) async for chunk in cast(AsyncIterator[bytes], response)]
if stream
else [cast(Mapping[str, object], response)]
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert events[-1]["type"] in ("message", "message_stop")
def test_sync_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
response: Final = LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
custom_llm_provider="openai",
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
@pytest.mark.asyncio
async def test_openai_turn_with_num_retries_sends_the_rejected_reasoning_only_once(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
def test_sync_openai_turn_with_global_num_retries_sends_the_rejected_reasoning_once_and_keeps_the_setting(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "num_retries", 2)
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert litellm.num_retries == 2
_SERVER_ERROR: Final = {
"error": {"message": "The server had an error while processing your request.", "type": "server_error"}
}
@pytest.mark.asyncio
async def test_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream_after(
respx_mock,
monkeypatch,
(httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
False,
)
response: Final = await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
def test_sync_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream_after(
respx_mock,
monkeypatch,
(httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
False,
)
response: Final = litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
class _FailureLog(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in ("encrypted content", "organization") if phrase in error), error),
)
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
pytest.param(httpx.Response(200, json=TEXT_RESPONSE), (), id="served"),
pytest.param(httpx.Response(400, json=UNVERIFIED_ORGANIZATION), ("organization",), id="rejected"),
)
def _rejected_then(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sent_again: httpx.Response) -> None:
monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
litellm.in_memory_llm_clients_cache.flush_cache()
respx_mock.post("https://api.openai.com/v1/responses").mock(
side_effect=[httpx.Response(400, json=REJECTED_REPLAYED_REASONING), sent_again]
)
@pytest.mark.asyncio
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
async def test_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
_rejected_then(respx_mock, monkeypatch, sent_again)
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert failure_log.rejections == logged_rejections
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
def test_sync_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
_rejected_then(respx_mock, monkeypatch, sent_again)
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert failure_log.rejections == logged_rejections
@pytest.mark.asyncio
@pytest.mark.parametrize(
"replayed_reasoning, rejection",
[
pytest.param(False, REJECTED_REPLAYED_REASONING, id="no_replayed_reasoning"),
pytest.param(True, UNVERIFIED_ORGANIZATION, id="unrelated_rejection"),
],
)
async def test_openai_turn_is_not_sent_again_for_other_rejections(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
replayed_reasoning: bool,
rejection: Mapping[str, object],
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=rejection), False)
with pytest.raises(litellm.BadRequestError):
await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=replayed_reasoning),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
custom_llm_provider="openai",
)
assert upstream.call_count == 1

15
tests/unit/test_main.py
View file

@ -8,7 +8,7 @@ import logging
import os
import threading
import urllib.parse
from collections.abc import Callable, Mapping
from collections.abc import Awaitable, Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime
@ -9395,6 +9395,19 @@ async def test_responses_with_retries(sync_mode):
assert mock_responses.call_args.kwargs["max_retries"] == 0
@pytest.mark.parametrize("retry_function", [acompletion_with_retries, aresponses_with_retries])
@pytest.mark.asyncio
async def test_async_retry_helpers_stop_when_the_call_is_cancelled(
retry_function: Callable[..., Awaitable[object]],
) -> None:
upstream: Final = AsyncMock(side_effect=[asyncio.CancelledError(), MagicMock()])
with pytest.raises(asyncio.CancelledError):
await retry_function(model="gpt-4o", num_retries=3, original_function=upstream)
assert upstream.await_count == 1
def test_azure_embedding_exceptions():
with pytest.raises(Exception, match="Mock error") as exc_info:
litellm.embedding(