scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

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

Browse source
This commit is contained in:
mateo-berri 2026-11-06 12:35:39 -08:00
parent 3bba28012a
commit 10c953d855
8 changed files with 418 additions and 81 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -485,6 +485,7 @@ def _as_chat_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")
@ -497,21 +498,33 @@ def _with_stream_block_index(
signature: Final = block.get("signature")
return ChatCompletionThinkingBlock(
type="thinking",
thinking=thinking if isinstance(thinking, str) else "",
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: object,
output_index: int | None,
summary_streamed: bool,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
index: Final = output_index if isinstance(output_index, int) and not isinstance(output_index, bool) else 0
return tuple(_with_stream_block_index(block, index) for block in thinking_blocks_from_reasoning_items((item,)))
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]) -> ModelResponseStream | None:
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
@ -525,7 +538,8 @@ def _reasoning_done_stream_chunk(chunk: Mapping[str, object]) -> ModelResponseSt
item: Final = as_str_mapping(chunk.get("item"))
if item is None or item.get("type") != "reasoning":
return None
signed_blocks: Final = _reasoning_done_thinking_blocks(item, chunk.get("output_index"))
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(
@ -593,7 +607,6 @@ def _fold_choices_into_one(
*chain.from_iterable(_message_reasoning_items(m) for m in messages),
*trailing_reasoning_items,
)
encrypted_thinking_blocks: Final = thinking_blocks_from_reasoning_items(reasoning_items)
return Choices(
message=Message(
role=messages[0].role if messages else "assistant",
@ -602,7 +615,6 @@ def _fold_choices_into_one(
tool_calls=list(tool_calls) or None,
reasoning_content=" ".join(value for value in reasoning_contents if value) or None,
reasoning_items=_as_chat_reasoning_items(reasoning_items),
thinking_blocks=list(encrypted_thinking_blocks) or None,
provider_specific_fields=_tool_search_provider_fields(tool_search_items),
),
finish_reason="tool_calls" if tool_calls else "stop",
@ -1496,12 +1508,10 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
for summary_block in reasoning_item["summary"]
if summary_block.get("text")
)
encrypted_thinking_blocks: Final = thinking_blocks_from_reasoning_items(reasoning_items)
message: Final = Message(
content=None if tool_search_items else "",
reasoning_content=reasoning_content if reasoning_content else None,
reasoning_items=_as_chat_reasoning_items(reasoning_items),
thinking_blocks=list(encrypted_thinking_blocks) or None,
provider_specific_fields=_tool_search_provider_fields(tool_search_items),
)
return Choices(message=message, finish_reason=finish_reason, index=0)
@ -2093,6 +2103,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
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"]
@ -2426,7 +2437,7 @@ 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)
reasoning_chunk: Final = _reasoning_done_stream_chunk(chunk, self._summary_streamed_outputs)
translated: Final = (
self._empty_content_chunk()
if self._is_text_delta_after_structured_answer(chunk)
@ -2444,6 +2455,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:

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

@ -282,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,
*,

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

@ -1,5 +1,6 @@
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping, Sequence
from dataclasses import dataclass
from functools import partial
from typing import (
TYPE_CHECKING,
Final,
@ -15,11 +16,13 @@ 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.litellm_logging import Logging as LiteLLMLoggingObject
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import messages_carry_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,
)
@ -53,6 +56,7 @@ 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 +74,68 @@ class _CompletionKwargs(TypedDict, total=False, extra_items=object):
custom_llm_provider: str
_ANTHROPIC_MESSAGES: Final = TypeAdapter(list[dict[str, object]])
_PrepareReplay: TypeAlias = "Callable[[_AnthropicMessages], _CompletionKwargs]"
@dataclass(frozen=True, slots=True)
class _ChatTargetReasoning:
effort: str
summary: str | None = None
def _messages_without_rejected_reasoning(
messages: _AnthropicMessages,
error: litellm.BadRequestError,
) -> _AnthropicMessages | None:
error_message: Final = str(error)
rejected: Final = "invalid_encrypted_content" in error_message or "could not be verified" in error_message
if not rejected or not messages_carry_encrypted_reasoning(messages):
return None
verbose_logger.warning("Backend rejected replayed encrypted reasoning, retrying once without it: %s", error)
return _ANTHROPIC_MESSAGES.validate_python(strip_encrypted_reasoning_blocks_from_anthropic_messages(messages))
async def _acompletion_without_rejected_reasoning(
completion_kwargs: _CompletionKwargs,
messages: _AnthropicMessages,
prepare_replay: _PrepareReplay,
) -> tuple[_CompletionKwargs, "ChatResult"]:
try:
return completion_kwargs, await litellm.acompletion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(messages, error)
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]"]:
try:
return completion_kwargs, litellm.completion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(messages, error)
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":
@ -804,12 +854,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 +871,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 +1000,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 +1017,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:

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

@ -27,9 +27,8 @@ import litellm
from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.exceptions import ContextWindowExceededError, MidStreamFallbackError
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items, attribute_or_none
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items
from litellm.litellm_core_utils.hidden_params import set_hidden_params
from litellm.litellm_core_utils.prompt_templates.reasoning_utils import thinking_blocks_from_reasoning_items
from litellm.llms.anthropic.messages.safeguards import StreamedSafeguardResults
from litellm.llms.anthropic.messages.utils import anthropic_sse_frame
from litellm.llms.base_llm.chat.transformation import BaseLLMException
@ -93,10 +92,6 @@ def _optional_attr(obj: object, name: str) -> object:
return getattr(obj, name, None)
def _delta_has_encrypted_reasoning(delta: object) -> bool:
return bool(thinking_blocks_from_reasoning_items(attribute_or_none(delta, "reasoning_items")))
def _optional_attr_sequence(obj: object, name: str) -> Sequence[object]:
value: Final = getattr(obj, name, None)
return value if value else ()
@ -1523,8 +1518,6 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
for b in attribute_items(delta, "thinking_blocks")
):
return False
if _delta_has_encrypted_reasoning(delta):
return False
return True
def _record_anthropic_stop_fields(self, delta: object) -> None:

64
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 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 (
@ -259,14 +260,32 @@ 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)
@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
_NO_ROUTE_INPUTS: Final = ChatRouteInputs()
def _keeps_encrypted_reasoning_for_responses(
model: str | None,
custom_llm_provider: str | None,
deployment_model_info: Mapping[str, object] | None,
route: ChatRouteInputs,
) -> bool:
from litellm.responses.bridge_check import chat_completion_ends_in_responses_api
return chat_completion_ends_in_responses_api(model, custom_llm_provider, deployment_model_info)
return chat_completion_ends_in_responses_api(
model,
custom_llm_provider,
deployment_model_info,
tools=route.tools,
reasoning_effort=route.reasoning_effort,
api_base=route.api_base,
)
def _anthropic_thinking_blocks(source: object) -> tuple[Mapping[str, object], ...]:
@ -276,6 +295,10 @@ def _anthropic_thinking_blocks(source: object) -> tuple[Mapping[str, object], ..
)
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:
eager_input_streaming: Final = eager_input_streaming_flag(tool)
if eager_input_streaming is None:
@ -545,11 +568,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 _keeps_encrypted_reasoning_for_responses(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(
@ -1349,27 +1373,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(
@ -1396,6 +1408,22 @@ 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,
),
)
## 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,
@ -1805,7 +1833,7 @@ class LiteLLMAnthropicMessagesAdapter:
choice.delta
) is not None:
return "text", TextBlock(type="text", text="")
elif isinstance(choice, StreamingChoices) and (thinking_blocks := _anthropic_thinking_blocks(choice.delta)):
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")
@ -1855,7 +1883,7 @@ 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 (thinking_blocks := _anthropic_thinking_blocks(choice.delta)):
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 ""

10
litellm/responses/bridge_check.py
View file

@ -300,12 +300,17 @@ 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,
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, and a deployment pinned to ``mode: responses`` keep it.
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
@ -317,6 +322,9 @@ def chat_completion_ends_in_responses_api(
model_info, _resolved = responses_api_bridge_check(
model=resolved_model,
custom_llm_provider=provider,
tools=tools,
reasoning_effort=reasoning_effort,
api_base=api_base,
deployment_model_info=deployment_model_info,
)
return model_info.get("mode") == "responses"

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

@ -3915,6 +3915,74 @@ def test_reasoning_output_item_done_with_summary_signs_the_thinking_text():
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)
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)
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,
@ -3935,10 +4003,13 @@ def test_reasoning_output_item_done_without_encrypted_content_stays_empty():
assert getattr(result.choices[0].delta, "thinking_blocks", None) in (None, [])
def test_transform_response_copies_encrypted_reasoning_onto_thinking_blocks():
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
@ -3991,10 +4062,16 @@ def test_transform_response_copies_encrypted_reasoning_onto_thinking_blocks():
)
message: Final = result.choices[0].message
assert message.reasoning_items[0]["encrypted_content"] == encrypted
assert message.thinking_blocks[0]["type"] == "redacted_thinking"
assert message.thinking_blocks[0]["data"] == encrypted_reasoning_signature(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)
def test_streaming_function_call_tool_id_for_degenerate_call_id():
"""In streaming, Bedrock Mantle's function_call event carries a unique ``id``

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

@ -34,6 +34,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
@ -754,26 +755,50 @@ def test_translate_openai_reasoning_items_are_not_duplicated_when_thinking_block
assert [block["type"] for block in result] == ["redacted_thinking", "text"]
def test_streaming_chunk_with_only_reasoning_items_opens_a_signed_thinking_block():
encrypted: Final = "gAAAA_delta"
choices: Final = [
StreamingChoices(
finish_reason=None,
index=0,
delta=Delta(
reasoning_items=[
{"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
]
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"}},
],
),
)
]
block_type, start = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_chunk_to_anthropic_content_block(
choices=choices
AnthropicMessagesUserMessageParam(
role="user",
content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
),
],
tools=tools,
)
assert block_type == "redacted_thinking"
assert start["data"] == encrypted_reasoning_signature(encrypted)
@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
def test_translate_anthropic_messages_to_openai_sets_reasoning_content():
@ -5738,3 +5763,141 @@ 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"}}
)
_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,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_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(),
)
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"]
@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")
@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"),
],
)
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