mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-16 04:29:31 +00:00
fix(bedrock)!: replay redacted_thinking on Converse and keep its data as Claude's own string (#46532)
* fix(bedrock): send redacted_thinking back to Converse as reasoningContent.redactedContent Replaying an assistant turn that carried a redacted_thinking block to Bedrock Converse turned it into an empty unsigned reasoningText, which the assistant content builder then dropped as blank text. Bedrock returned 200 so callers never noticed. Map redacted_thinking to redactedContent in both the thinking_blocks and inline content paths of the sync and async builders * fix(bedrock): drop blank redacted_thinking data and keep non-str signatures signed on Converse replay * fix(bedrock): replay redacted_thinking to Converse only when the target is Claude Nova, Llama, DeepSeek and gpt-oss reject reasoningContent.redactedContent with a 400, so a Claude history replayed to them through a Router fallback or model switch keeps dropping the redacted block the way main does. Claude is resolved from the Bedrock base model of the target, and an application inference profile ARN counts as Claude when its deployment's base_model names a Claude model. Adds integration contracts for the replayed Converse request shape, stream and non-stream, with a tool call, and for non-Claude and ARN targets. * test(e2e): cover Bedrock Converse replaying redacted_thinking live through the proxy Turn 1 triggers redacted thinking on Claude Sonnet 4.6, turn 2 replays it, and a live edge between the proxy and Bedrock checks the replayed Converse body carries the same reasoningContent.redactedContent. Covers nonstream, stream and a tool call round trip * fix(bedrock): validate litellm_params before reading the deployment base_model litellm_params arrives as an untyped dict, so passing it straight into the typed helper grew reportUnknownArgumentType. Validating it into Mapping[str, object] first keeps the read typed without growing the type-check gate * fix(bedrock): wrap non-Converse redacted_thinking data and let a named model id win over base_model * test(bedrock): cover redacted_thinking minted outside Converse, model_id and the sync Router base_model path * fix(bedrock): match bare claude- base_model names without reading the untyped cost map * fix(bedrock): keep redacted_thinking data as Claude's own string on every route Converse hands redactedContent back base64-wrapped, and that wrapper used to leak out as data, so a Converse turn replayed to Invoke or the Anthropic API was rejected. Converse replies now decode it, Converse requests wrap it exactly once, and the Anthropic-shaped builders unwrap histories saved with the old wrapping. Responses API encrypted reasoning is no longer sent to Claude on Converse, which rejected it with a 400
This commit is contained in:
ryan-crabbe-berri
GitHub
parent
ff8b5d8324
commit
e889d9d653
18 changed files with 1511 additions and 94 deletions
|
|
@ -2,6 +2,8 @@
|
|||
Common utility functions used for translating messages across providers
|
||||
"""
|
||||
|
||||
import base64
|
||||
import binascii
|
||||
import io
|
||||
import json
|
||||
import mimetypes
|
||||
|
|
@ -2132,6 +2134,32 @@ def is_unsignable_thinking_block(block: object) -> bool:
|
|||
return not (isinstance(thinking_text, str) and len(thinking_text.strip()) > 0)
|
||||
|
||||
|
||||
def native_redacted_thinking_data(data: str) -> str:
|
||||
"""Claude's own `data` string for a redacted_thinking block.
|
||||
|
||||
Bedrock Converse carries it base64-wrapped as `redactedContent`, and LiteLLM passed that
|
||||
wrapper through as `data` until it started decoding Converse replies, so histories saved
|
||||
before then still hold it. Claude's own value never decodes twice, since its first byte
|
||||
decodes to a protobuf tag outside the base64 alphabet.
|
||||
"""
|
||||
return _converse_wrapped_payload(data) or data
|
||||
|
||||
|
||||
def converse_redacted_content(data: str) -> str:
|
||||
"""The `redactedContent` Bedrock Converse expects for a redacted_thinking block's `data`."""
|
||||
return base64.b64encode(native_redacted_thinking_data(data).encode()).decode()
|
||||
|
||||
|
||||
def _converse_wrapped_payload(data: str) -> str | None:
|
||||
try:
|
||||
payload: Final = base64.b64decode(data, validate=True).decode("ascii")
|
||||
_ = base64.b64decode(payload, validate=True)
|
||||
except (binascii.Error, ValueError):
|
||||
return None
|
||||
else:
|
||||
return payload
|
||||
|
||||
|
||||
def strip_encrypted_reasoning_from_messages(
|
||||
messages: object,
|
||||
*,
|
||||
|
|
|
|||
|
|
@ -54,10 +54,13 @@ from litellm.types.llms.vertex_ai import PartType as VertexPartType
|
|||
from litellm.types.utils import GenericImageParsingChunk
|
||||
|
||||
from .common_utils import (
|
||||
converse_redacted_content,
|
||||
convert_content_list_to_str,
|
||||
infer_content_type_from_url_and_content,
|
||||
is_encrypted_reasoning_block,
|
||||
is_non_content_values_set,
|
||||
is_unsignable_thinking_block,
|
||||
native_redacted_thinking_data,
|
||||
parse_tool_call_arguments,
|
||||
salvage_concatenated_tool_arguments,
|
||||
)
|
||||
|
|
@ -2486,10 +2489,21 @@ def sanitize_messages_for_tool_calling(
|
|||
return sanitized_messages
|
||||
|
||||
|
||||
def _drop_unsignable_thinking_blocks(
|
||||
def _replayable_thinking_blocks(
|
||||
thinking_blocks: list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock],
|
||||
) -> list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock]:
|
||||
return [block for block in thinking_blocks if not is_unsignable_thinking_block(block)]
|
||||
return [_with_native_redacted_data(block) for block in thinking_blocks if not is_unsignable_thinking_block(block)]
|
||||
|
||||
|
||||
def _with_native_redacted_data(
|
||||
block: ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
|
||||
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock:
|
||||
match block:
|
||||
case {"type": "redacted_thinking", "data": str() as data}:
|
||||
native: Final[ChatCompletionRedactedThinkingBlock] = {**block, "data": native_redacted_thinking_data(data)}
|
||||
return native
|
||||
case _:
|
||||
return block
|
||||
|
||||
|
||||
_AnthropicMessageList: TypeAlias = list[AllAnthropicPassThroughMessageValues]
|
||||
|
|
@ -2679,7 +2693,7 @@ def anthropic_messages_pt(
|
|||
|
||||
_raw_thinking_blocks = assistant_content_block.get("thinking_blocks", None)
|
||||
thinking_blocks = (
|
||||
_drop_unsignable_thinking_blocks(_raw_thinking_blocks) if _raw_thinking_blocks is not None else None
|
||||
_replayable_thinking_blocks(_raw_thinking_blocks) if _raw_thinking_blocks is not None else None
|
||||
)
|
||||
|
||||
# Check if tool_calls contain server tool calls (web search, etc.)
|
||||
|
|
@ -4523,9 +4537,13 @@ class BedrockConverseMessagesProcessor:
|
|||
llm_provider: str,
|
||||
user_continue_message: ChatCompletionUserMessage | None = None,
|
||||
assistant_continue_message: str | ChatCompletionAssistantMessage | None = None,
|
||||
base_model: str | None = None,
|
||||
) -> list[BedrockMessageBlock]:
|
||||
from litellm.llms.bedrock.common_utils import bedrock_converse_accepts_redacted_thinking
|
||||
|
||||
contents: list[BedrockMessageBlock] = []
|
||||
msg_i = 0
|
||||
to_claude: Final = bedrock_converse_accepts_redacted_thinking(model, base_model)
|
||||
|
||||
messages = BedrockConverseMessagesProcessor._initial_message_setup(
|
||||
messages, model, llm_provider, user_continue_message
|
||||
|
|
@ -4670,15 +4688,12 @@ class BedrockConverseMessagesProcessor:
|
|||
assistant_continue_message=assistant_continue_message,
|
||||
)
|
||||
_assistant_content = assistant_message_block.get("content", None)
|
||||
thinking_blocks = cast(
|
||||
list[ChatCompletionThinkingBlock] | None,
|
||||
assistant_message_block.get("thinking_blocks"),
|
||||
)
|
||||
thinking_blocks = assistant_message_block.get("thinking_blocks")
|
||||
|
||||
if thinking_blocks is not None:
|
||||
converted_thinking_blocks = (
|
||||
BedrockConverseMessagesProcessor.translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
thinking_blocks
|
||||
thinking_blocks, to_claude=to_claude
|
||||
)
|
||||
)
|
||||
assistant_content = BedrockConverseMessagesProcessor.add_thinking_blocks_to_assistant_content(
|
||||
|
|
@ -4690,9 +4705,9 @@ class BedrockConverseMessagesProcessor:
|
|||
assistants_parts: list[BedrockContentBlock] = []
|
||||
for element in _assistant_content:
|
||||
if isinstance(element, dict):
|
||||
if element["type"] == "thinking":
|
||||
if element["type"] == "thinking" or element["type"] == "redacted_thinking":
|
||||
thinking_block = BedrockConverseMessagesProcessor.translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
thinking_blocks=[element]
|
||||
thinking_blocks=[element], to_claude=to_claude
|
||||
)
|
||||
assistants_parts = (
|
||||
BedrockConverseMessagesProcessor.add_thinking_blocks_to_assistant_content(
|
||||
|
|
@ -4756,23 +4771,49 @@ class BedrockConverseMessagesProcessor:
|
|||
|
||||
@staticmethod
|
||||
def translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
thinking_blocks: list[ChatCompletionThinkingBlock],
|
||||
thinking_blocks: Sequence[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock],
|
||||
*,
|
||||
to_claude: bool = False,
|
||||
) -> list[BedrockContentBlock]:
|
||||
reasoning_content_blocks: Final[list[BedrockContentBlock]] = []
|
||||
for thinking_block in thinking_blocks:
|
||||
reasoning_text = thinking_block.get("thinking")
|
||||
reasoning_signature = thinking_block.get("signature")
|
||||
text_block = BedrockConverseReasoningTextBlock(
|
||||
text=reasoning_text or "",
|
||||
return [
|
||||
BedrockContentBlock(
|
||||
reasoningContent=BedrockConverseMessagesProcessor._to_reasoning_content_block(
|
||||
thinking_block, to_claude=to_claude
|
||||
)
|
||||
)
|
||||
if reasoning_signature is not None:
|
||||
text_block["signature"] = reasoning_signature
|
||||
reasoning_content_block = BedrockConverseReasoningContentBlock(
|
||||
reasoningText=text_block,
|
||||
)
|
||||
bedrock_content_block = BedrockContentBlock(reasoningContent=reasoning_content_block)
|
||||
reasoning_content_blocks.append(bedrock_content_block)
|
||||
return reasoning_content_blocks
|
||||
for thinking_block in thinking_blocks
|
||||
if not (to_claude and is_encrypted_reasoning_block(thinking_block))
|
||||
]
|
||||
|
||||
@staticmethod
|
||||
def _to_reasoning_content_block(
|
||||
thinking_block: ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock,
|
||||
*,
|
||||
to_claude: bool,
|
||||
) -> BedrockConverseReasoningContentBlock:
|
||||
match thinking_block:
|
||||
case {"type": "redacted_thinking", "data": str() as data} if to_claude and data.strip():
|
||||
return BedrockConverseReasoningContentBlock(redactedContent=converse_redacted_content(data))
|
||||
case {"signature": signature} if signature is not None:
|
||||
return BedrockConverseReasoningContentBlock(
|
||||
reasoningText=BedrockConverseReasoningTextBlock(
|
||||
text=BedrockConverseMessagesProcessor._thinking_text(thinking_block), signature=signature
|
||||
)
|
||||
)
|
||||
case _:
|
||||
return BedrockConverseReasoningContentBlock(
|
||||
reasoningText=BedrockConverseReasoningTextBlock(
|
||||
text=BedrockConverseMessagesProcessor._thinking_text(thinking_block)
|
||||
)
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
def _thinking_text(thinking_block: ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock) -> str:
|
||||
match thinking_block:
|
||||
case {"thinking": str() as text}:
|
||||
return text
|
||||
case _:
|
||||
return ""
|
||||
|
||||
@staticmethod
|
||||
def process_file_message(message: ChatCompletionFileObject) -> BedrockContentBlock:
|
||||
|
|
@ -4890,6 +4931,7 @@ def _bedrock_converse_messages_pt(
|
|||
llm_provider: str,
|
||||
user_continue_message: ChatCompletionUserMessage | None = None,
|
||||
assistant_continue_message: str | ChatCompletionAssistantMessage | None = None,
|
||||
base_model: str | None = None,
|
||||
) -> list[BedrockMessageBlock]:
|
||||
"""
|
||||
Converts given messages from OpenAI format to Bedrock format
|
||||
|
|
@ -4898,9 +4940,11 @@ def _bedrock_converse_messages_pt(
|
|||
- Please ensure that function response turn comes immediately after a function call turn
|
||||
- Conversation blocks and tool result blocks cannot be provided in the same turn. Issue: https://github.com/BerriAI/litellm/issues/6053
|
||||
"""
|
||||
from litellm.llms.bedrock.common_utils import bedrock_converse_accepts_redacted_thinking
|
||||
|
||||
contents: list[BedrockMessageBlock] = []
|
||||
msg_i = 0
|
||||
to_claude: Final = bedrock_converse_accepts_redacted_thinking(model, base_model)
|
||||
|
||||
messages = BedrockConverseMessagesProcessor._initial_message_setup(
|
||||
messages, model, llm_provider, user_continue_message
|
||||
|
|
@ -5046,15 +5090,12 @@ def _bedrock_converse_messages_pt(
|
|||
assistant_continue_message=assistant_continue_message,
|
||||
)
|
||||
_assistant_content = assistant_message_block.get("content", None)
|
||||
thinking_blocks = cast(
|
||||
list[ChatCompletionThinkingBlock] | None,
|
||||
assistant_message_block.get("thinking_blocks"),
|
||||
)
|
||||
thinking_blocks = assistant_message_block.get("thinking_blocks")
|
||||
|
||||
if thinking_blocks is not None:
|
||||
converted_thinking_blocks = (
|
||||
BedrockConverseMessagesProcessor.translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
thinking_blocks
|
||||
thinking_blocks, to_claude=to_claude
|
||||
)
|
||||
)
|
||||
assistant_content = BedrockConverseMessagesProcessor.add_thinking_blocks_to_assistant_content(
|
||||
|
|
@ -5066,10 +5107,10 @@ def _bedrock_converse_messages_pt(
|
|||
assistants_parts: list[BedrockContentBlock] = []
|
||||
for element in _assistant_content:
|
||||
if isinstance(element, dict):
|
||||
if element["type"] == "thinking":
|
||||
if element["type"] == "thinking" or element["type"] == "redacted_thinking":
|
||||
thinking_block = (
|
||||
BedrockConverseMessagesProcessor.translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
thinking_blocks=[element]
|
||||
thinking_blocks=[element], to_claude=to_claude
|
||||
)
|
||||
)
|
||||
assistants_parts = (
|
||||
|
|
|
|||
|
|
@ -26,6 +26,7 @@ from litellm.exceptions import UnsupportedParamsError
|
|||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
get_file_ids_from_messages,
|
||||
is_encrypted_reasoning_block,
|
||||
native_redacted_thinking_data,
|
||||
)
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import (
|
||||
THOUGHT_SIGNATURE_SEPARATOR,
|
||||
|
|
@ -1641,29 +1642,43 @@ def strip_thinking_blocks_from_anthropic_messages(messages: Sequence[object]) ->
|
|||
return out
|
||||
|
||||
|
||||
def _without_encrypted_reasoning_blocks(message: dict) -> dict | None: # mutable-ok: Anthropic message payload shape
|
||||
def _replayable_message(message: dict) -> dict | None: # mutable-ok: Anthropic message payload shape
|
||||
if not isinstance(message, Mapping):
|
||||
return message
|
||||
content: Final = message.get("content")
|
||||
if not isinstance(content, list):
|
||||
return message
|
||||
kept: Final = [b for b in content if not is_encrypted_reasoning_block(b)]
|
||||
if len(kept) == len(content):
|
||||
kept: Final = [_with_native_redacted_data(b) for b in content if not is_encrypted_reasoning_block(b)]
|
||||
if kept == content:
|
||||
return message
|
||||
if not kept:
|
||||
return None
|
||||
return {**message, "content": kept}
|
||||
|
||||
|
||||
_CONTENT_BLOCK: Final = TypeAdapter(dict[str, object])
|
||||
|
||||
|
||||
def _with_native_redacted_data(block: object) -> object:
|
||||
if not isinstance(block, Mapping):
|
||||
return block
|
||||
match _CONTENT_BLOCK.validate_python(block):
|
||||
case {"type": "redacted_thinking", "data": str() as data} as redacted:
|
||||
return {**redacted, "data": native_redacted_thinking_data(data)}
|
||||
case _:
|
||||
return block
|
||||
|
||||
|
||||
def strip_encrypted_reasoning_blocks_from_anthropic_messages(
|
||||
messages: Sequence[dict], # mutable-ok: Anthropic message payload shape
|
||||
) -> list[dict]: # mutable-ok: AnthropicMessagesRequest.messages is typed list[dict]
|
||||
"""
|
||||
Drop thinking / redacted_thinking blocks that carry another provider's encrypted
|
||||
reasoning (a turn the Responses API bridge served) before the request reaches
|
||||
Anthropic, which cannot verify them. Anthropic's own signed blocks are kept.
|
||||
Anthropic, which cannot verify them. Anthropic's own signed blocks are kept, with a
|
||||
redacted block's Bedrock Converse-wrapped `data` unwrapped back to Claude's own.
|
||||
"""
|
||||
stripped: Final = (_without_encrypted_reasoning_blocks(m) for m in messages)
|
||||
stripped: Final = (_replayable_message(m) for m in messages)
|
||||
return [m for m in stripped if m is not None]
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ from litellm.litellm_core_utils.core_helpers import (
|
|||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
drop_lookaround_regex_patterns,
|
||||
native_redacted_thinking_data,
|
||||
parse_content_for_reasoning,
|
||||
tool_with_sanitized_parameters,
|
||||
)
|
||||
|
|
@ -136,6 +137,7 @@ UNSUPPORTED_BEDROCK_CONVERSE_BETA_PATTERNS: Final = [
|
|||
|
||||
|
||||
_TOOLS_AS_SENT: Final = TypeAdapter(tuple[Mapping[str, object], ...])
|
||||
_DEPLOYMENT_PARAMS: Final = TypeAdapter(Mapping[str, object])
|
||||
|
||||
|
||||
def _tools_the_model_accepts(
|
||||
|
|
@ -2045,6 +2047,7 @@ class AmazonConverseConfig(BaseConfig):
|
|||
model=model,
|
||||
llm_provider="bedrock_converse",
|
||||
user_continue_message=litellm_params.pop("user_continue_message", None),
|
||||
base_model=self._deployment_base_model(_DEPLOYMENT_PARAMS.validate_python(litellm_params)),
|
||||
),
|
||||
model,
|
||||
)
|
||||
|
|
@ -2061,6 +2064,11 @@ class AmazonConverseConfig(BaseConfig):
|
|||
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
def _deployment_base_model(litellm_params: Mapping[str, object]) -> str | None:
|
||||
base_model: Final = litellm_params.get("base_model")
|
||||
return base_model if isinstance(base_model, str) else None
|
||||
|
||||
def transform_request(
|
||||
self,
|
||||
model: str,
|
||||
|
|
@ -2113,6 +2121,7 @@ class AmazonConverseConfig(BaseConfig):
|
|||
model=model,
|
||||
llm_provider="bedrock_converse",
|
||||
user_continue_message=litellm_params.pop("user_continue_message", None),
|
||||
base_model=self._deployment_base_model(_DEPLOYMENT_PARAMS.validate_python(litellm_params)),
|
||||
),
|
||||
model,
|
||||
)
|
||||
|
|
@ -2186,7 +2195,7 @@ class AmazonConverseConfig(BaseConfig):
|
|||
thinking_blocks_list.append(_thinking_block)
|
||||
elif "redactedContent" in block:
|
||||
_redacted_block = ChatCompletionRedactedThinkingBlock(
|
||||
type="redacted_thinking", data=block["redactedContent"]
|
||||
type="redacted_thinking", data=native_redacted_thinking_data(block["redactedContent"])
|
||||
)
|
||||
thinking_blocks_list.append(_redacted_block)
|
||||
return thinking_blocks_list
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
|
|||
from litellm.litellm_core_utils.core_helpers import map_finish_reason
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.optional_imports import ensure_optional_import
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import native_redacted_thinking_data
|
||||
from litellm.llms.anthropic.chat.handler import (
|
||||
ModelResponseIterator as AnthropicModelResponseIterator,
|
||||
)
|
||||
|
|
@ -532,7 +533,7 @@ class AWSEventStreamDecoder:
|
|||
_thinking_block["thinking"] = "" # consistent with anthropic response
|
||||
elif "redactedContent" in thinking_block:
|
||||
_thinking_block = ChatCompletionRedactedThinkingBlock(
|
||||
type="redacted_thinking", data=thinking_block["redactedContent"]
|
||||
type="redacted_thinking", data=native_redacted_thinking_data(thinking_block["redactedContent"])
|
||||
)
|
||||
if _thinking_block is not None:
|
||||
thinking_blocks_list.append(_thinking_block)
|
||||
|
|
|
|||
|
|
@ -1164,6 +1164,24 @@ def get_bedrock_base_model(model: str) -> str:
|
|||
return model
|
||||
|
||||
|
||||
def bedrock_converse_accepts_redacted_thinking(model: str, base_model: str | None) -> bool:
|
||||
"""
|
||||
Whether ``reasoningContent.redactedContent`` may be replayed to this Converse target.
|
||||
|
||||
Only Claude takes it back. Nova, Llama, DeepSeek and gpt-oss 400 on it (checked live in us-west-2 on
|
||||
2026-10-14), so every other target keeps dropping the block. A model id that names its model decides on
|
||||
its own. An ARN that hides its model, like an application inference profile, falls back to the
|
||||
deployment's ``base_model`` and counts as non-Claude without one.
|
||||
"""
|
||||
if base_model is None or "." in get_bedrock_base_model(model):
|
||||
return _is_claude(model)
|
||||
return _is_claude(base_model)
|
||||
|
||||
|
||||
def _is_claude(model: str) -> bool:
|
||||
return get_bedrock_base_model(model).startswith(("anthropic", "claude"))
|
||||
|
||||
|
||||
def bedrock_converse_supports_parallel_tool_use_config(model: str) -> bool:
|
||||
return any(
|
||||
(litellm.model_cost.get(candidate) or {}).get("supports_parallel_tool_use_config") is True
|
||||
|
|
|
|||
|
|
@ -0,0 +1,487 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import queue
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from itertools import chain
|
||||
from types import MappingProxyType
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from _pytest.mark.structures import ParameterSet
|
||||
from e2e_config import PROVIDER_EDGE_ADVERTISE_HOST, PROVIDER_EDGE_BIND_HOST, unique_marker
|
||||
from e2e_http import unwrap
|
||||
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
|
||||
from lifecycle import ResourceManager
|
||||
from models import (
|
||||
AnthropicMessagesBody,
|
||||
ChatAssistantTurn,
|
||||
ChatBody,
|
||||
ChatMessage,
|
||||
ChatTool,
|
||||
ChatToolFunction,
|
||||
ChatToolResultTurn,
|
||||
LiteLLMParamsBody,
|
||||
OutMessage,
|
||||
ThinkingBlock,
|
||||
ThinkingParam,
|
||||
)
|
||||
from provider_cache import RequestSigner
|
||||
from provider_edge import LiveEdge, start_provider_edge
|
||||
from provider_edge_bedrock import bedrock_signer
|
||||
from proxy_client import ProxyClient
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
pytestmark = pytest.mark.e2e
|
||||
|
||||
# Sonnet 4.6 is the newest Claude on Bedrock that still honors Anthropic's redacted-thinking test string
|
||||
# (Sonnet 5.5 answers it with plain thinking, checked 2026-10-14). Source for the string:
|
||||
# https://docs.anthropic.com/en/docs/build-with-claude/extended-thinking#thinking-redaction
|
||||
BACKEND: Final = "bedrock/converse/us.anthropic.claude-sonnet-4-6"
|
||||
INVOKE_BACKEND: Final = "bedrock/invoke/us.anthropic.claude-sonnet-4-6"
|
||||
ANTHROPIC_BACKEND: Final = "anthropic/claude-sonnet-4-6"
|
||||
OPENAI_BACKEND: Final = "openai/gpt-5.6"
|
||||
ENCRYPTED_REASONING_PREFIX: Final = "litellm_encrypted_reasoning:"
|
||||
REDACTED_THINKING_TRIGGER: Final = (
|
||||
"ANTHROPIC_MAGIC_STRING_TRIGGER_REDACTED_THINKING_46C9A13E193C177646C7398A98432ECCCE4C1253D5E2D82641AC0E52CC2876CB"
|
||||
)
|
||||
REGION: Final = "us-west-2"
|
||||
EDGE_MOUNT: Final = f"bedrock/{REGION}"
|
||||
ANTHROPIC_MOUNT: Final = "anthropic"
|
||||
THINKING: Final = ThinkingParam(type="enabled", budget_tokens=1024)
|
||||
OPENAI_THINKING: Final = ThinkingParam(type="enabled", budget_tokens=4096)
|
||||
FOLLOW_UP: Final = "Reply with the single word ok."
|
||||
WEATHER_TOOL: Final = ChatTool(
|
||||
function=ChatToolFunction(
|
||||
name="get_weather",
|
||||
description="Get the current weather for a city",
|
||||
parameters={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
class _Frozen(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
|
||||
class _ReasoningContent(_Frozen):
|
||||
redactedContent: str | None = None
|
||||
|
||||
|
||||
class _ConverseContentBlock(_Frozen):
|
||||
reasoningContent: _ReasoningContent | None = None
|
||||
|
||||
|
||||
class _ConverseMessage(_Frozen):
|
||||
role: str
|
||||
content: tuple[_ConverseContentBlock, ...]
|
||||
|
||||
|
||||
class _ConverseBody(_Frozen):
|
||||
messages: tuple[_ConverseMessage, ...]
|
||||
|
||||
|
||||
class _AnthropicWireBlock(_Frozen):
|
||||
type: str
|
||||
data: str | None = None
|
||||
|
||||
|
||||
class _AnthropicWireMessage(_Frozen):
|
||||
role: str
|
||||
content: str | tuple[_AnthropicWireBlock, ...]
|
||||
|
||||
|
||||
class _AnthropicWireBody(_Frozen):
|
||||
messages: tuple[_AnthropicWireMessage, ...]
|
||||
|
||||
|
||||
class _Delta(_Frozen):
|
||||
content: str | None = None
|
||||
thinking_blocks: tuple[ThinkingBlock, ...] | None = None
|
||||
|
||||
|
||||
class _StreamChoice(_Frozen):
|
||||
delta: _Delta
|
||||
|
||||
|
||||
class _StreamChunk(_Frozen):
|
||||
choices: tuple[_StreamChoice, ...] = ()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _WireCapture:
|
||||
"""Every (url, body) the proxy sent upstream through the edges this test started"""
|
||||
|
||||
_requests: queue.SimpleQueue[tuple[str, bytes]] = field(default_factory=queue.SimpleQueue)
|
||||
|
||||
def observe(self, url: str, headers: Mapping[str, str], body: bytes | None) -> None:
|
||||
if body is not None:
|
||||
self._requests.put((url, body))
|
||||
|
||||
def last_body(self, path_marker: str) -> bytes:
|
||||
sent: Final = tuple(self._requests.get_nowait() for _ in range(self._requests.qsize()))
|
||||
bodies: Final = tuple(body for url, body in sent if path_marker in url)
|
||||
assert bodies, f"the proxy sent no {path_marker} request through the edge, only {[url for url, _ in sent]}"
|
||||
return bodies[-1]
|
||||
|
||||
|
||||
def _converse_reasoning(body: bytes) -> tuple[_ReasoningContent, ...]:
|
||||
assistant: Final = (
|
||||
message for message in _ConverseBody.model_validate_json(body).messages if message.role == "assistant"
|
||||
)
|
||||
blocks: Final = (block.reasoningContent for block in chain.from_iterable(m.content for m in assistant))
|
||||
return tuple(block for block in blocks if block is not None)
|
||||
|
||||
|
||||
def _converse_redacted(body: bytes) -> tuple[str, ...]:
|
||||
return tuple(block.redactedContent for block in _converse_reasoning(body) if block.redactedContent)
|
||||
|
||||
|
||||
def _anthropic_redacted(body: bytes) -> tuple[str, ...]:
|
||||
assistant: Final = (
|
||||
message.content
|
||||
for message in _AnthropicWireBody.model_validate_json(body).messages
|
||||
if message.role == "assistant" and not isinstance(message.content, str)
|
||||
)
|
||||
return tuple(block.data or "" for block in chain.from_iterable(assistant) if block.type == "redacted_thinking")
|
||||
|
||||
|
||||
def _b64(data: str) -> str:
|
||||
return base64.b64encode(data.encode()).decode()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _ClaudeRoute:
|
||||
"""One way the proxy reaches Claude, with how that route carries a redacted_thinking block's ``data``"""
|
||||
|
||||
backend: str
|
||||
mount: str
|
||||
path_marker: str
|
||||
provider: Provider
|
||||
replayed: Callable[[bytes], tuple[str, ...]]
|
||||
on_the_wire: Callable[[str], str]
|
||||
|
||||
|
||||
CONVERSE: Final = _ClaudeRoute(BACKEND, EDGE_MOUNT, "/converse", Provider.BEDROCK, _converse_redacted, _b64)
|
||||
INVOKE: Final = _ClaudeRoute(INVOKE_BACKEND, EDGE_MOUNT, "/invoke", Provider.BEDROCK, _anthropic_redacted, str)
|
||||
ANTHROPIC: Final = _ClaudeRoute(
|
||||
ANTHROPIC_BACKEND, ANTHROPIC_MOUNT, "/v1/messages", Provider.ANTHROPIC, _anthropic_redacted, str
|
||||
)
|
||||
CLAUDE_ROUTES: Final = MappingProxyType({"converse": CONVERSE, "invoke": INVOKE, "anthropic": ANTHROPIC})
|
||||
|
||||
|
||||
def _start_edge(
|
||||
resources: ResourceManager, capture: _WireCapture, mount: str, upstream: str, sign: RequestSigner | None
|
||||
) -> str:
|
||||
edge: Final = start_provider_edge(
|
||||
LiveEdge(observe_request=capture.observe, sign=sign),
|
||||
mounts=MappingProxyType({mount: upstream}),
|
||||
bind_host=PROVIDER_EDGE_BIND_HOST,
|
||||
advertise_host=PROVIDER_EDGE_ADVERTISE_HOST,
|
||||
)
|
||||
resources.defer(edge.shutdown)
|
||||
return edge.edge.api_base(mount)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Edges:
|
||||
bedrock: str
|
||||
anthropic: str
|
||||
capture: _WireCapture
|
||||
|
||||
|
||||
def _edges(resources: ResourceManager) -> _Edges:
|
||||
capture: Final = _WireCapture()
|
||||
return _Edges(
|
||||
bedrock=_start_edge(
|
||||
resources, capture, EDGE_MOUNT, f"https://bedrock-runtime.{REGION}.amazonaws.com", bedrock_signer(REGION)
|
||||
),
|
||||
anthropic=_start_edge(resources, capture, ANTHROPIC_MOUNT, "https://api.anthropic.com", None),
|
||||
capture=capture,
|
||||
)
|
||||
|
||||
|
||||
def _create(proxy: ProxyClient, resources: ResourceManager, params: LiteLLMParamsBody) -> str:
|
||||
model: Final = f"e2e-redacted-thinking-{unique_marker()}"
|
||||
model_id: Final = proxy.create_model(model, params)
|
||||
resources.defer(lambda: proxy.delete_model(model_id))
|
||||
return model
|
||||
|
||||
|
||||
def _deploy(proxy: ProxyClient, resources: ResourceManager, edges: _Edges, route: _ClaudeRoute) -> str:
|
||||
if route.provider is Provider.ANTHROPIC:
|
||||
return _create(
|
||||
proxy,
|
||||
resources,
|
||||
LiteLLMParamsBody(model=route.backend, api_base=edges.anthropic, api_key="os.environ/ANTHROPIC_API_KEY"),
|
||||
)
|
||||
return _create(
|
||||
proxy,
|
||||
resources,
|
||||
LiteLLMParamsBody(
|
||||
model=route.backend,
|
||||
api_base=edges.bedrock,
|
||||
aws_access_key_id="os.environ/AWS_ACCESS_KEY_ID",
|
||||
aws_secret_access_key="os.environ/AWS_SECRET_ACCESS_KEY",
|
||||
aws_region_name=REGION,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _chat(proxy: ProxyClient, key: str, body: ChatBody) -> OutMessage:
|
||||
response: Final = unwrap(proxy.chat(key, body))
|
||||
message: Final = response.choices[0].message if response.choices else None
|
||||
assert message is not None, f"chat returned no message: {response}"
|
||||
return message
|
||||
|
||||
|
||||
def _chat_stream(proxy: ProxyClient, key: str, body: ChatBody) -> OutMessage:
|
||||
result: Final = proxy.chat_stream(key, body.model_copy(update={"stream": True}))
|
||||
assert result.ok and result.is_streaming, f"stream was not established: {result}"
|
||||
assert result.stream_error is None, f"stream carried an error event: {result.stream_error}"
|
||||
chunks: Final = tuple(_StreamChunk.model_validate_json(event) for event in result.stream_events)
|
||||
deltas: Final = tuple(choice.delta for choice in chain.from_iterable(chunk.choices for chunk in chunks))
|
||||
return OutMessage(
|
||||
content="".join(delta.content or "" for delta in deltas),
|
||||
thinking_blocks=list(chain.from_iterable(delta.thinking_blocks or () for delta in deltas)),
|
||||
)
|
||||
|
||||
|
||||
def _redacted_data(message: OutMessage) -> tuple[str, ...]:
|
||||
data: Final = tuple(
|
||||
block.data or "" for block in message.thinking_blocks or () if block.type == "redacted_thinking"
|
||||
)
|
||||
assert data and all(data), f"turn 1 came back without redacted thinking: {message.thinking_blocks}"
|
||||
return data
|
||||
|
||||
|
||||
def _replay(model: str, question: ChatMessage, first: OutMessage) -> ChatBody:
|
||||
return ChatBody(
|
||||
model=model,
|
||||
thinking=THINKING,
|
||||
max_tokens=2048,
|
||||
messages=[
|
||||
question,
|
||||
ChatAssistantTurn(content=first.content, thinking_blocks=first.thinking_blocks),
|
||||
ChatMessage(role="user", content=FOLLOW_UP),
|
||||
],
|
||||
)
|
||||
|
||||
|
||||
def _route_pair(origin: str, destination: str) -> ParameterSet:
|
||||
routes: Final = (CLAUDE_ROUTES[origin], CLAUDE_ROUTES[destination])
|
||||
return pytest.param(
|
||||
origin,
|
||||
destination,
|
||||
marks=meta(
|
||||
Subject(
|
||||
domain=Domain.LLM_TRANSLATION,
|
||||
route=Route.CHAT_COMPLETIONS,
|
||||
providers=tuple(route.provider for route in routes),
|
||||
models=tuple(route.backend for route in routes),
|
||||
capabilities=(Capability.REASONING,),
|
||||
mode=Mode.NONSTREAM,
|
||||
)
|
||||
),
|
||||
id=f"{origin}-to-{destination}",
|
||||
)
|
||||
|
||||
|
||||
class TestBedrockConverseRedactedThinkingReplay:
|
||||
@pytest.mark.provider_edge_host
|
||||
@pytest.mark.parametrize(
|
||||
"stream",
|
||||
[
|
||||
pytest.param(
|
||||
False,
|
||||
marks=meta(
|
||||
Subject(
|
||||
domain=Domain.LLM_TRANSLATION,
|
||||
route=Route.CHAT_COMPLETIONS,
|
||||
providers=(Provider.BEDROCK,),
|
||||
models=(BACKEND,),
|
||||
capabilities=(Capability.REASONING,),
|
||||
mode=Mode.NONSTREAM,
|
||||
)
|
||||
),
|
||||
id="nonstream",
|
||||
),
|
||||
pytest.param(
|
||||
True,
|
||||
marks=meta(
|
||||
Subject(
|
||||
domain=Domain.LLM_TRANSLATION,
|
||||
route=Route.CHAT_COMPLETIONS,
|
||||
providers=(Provider.BEDROCK,),
|
||||
models=(BACKEND,),
|
||||
capabilities=(Capability.REASONING,),
|
||||
mode=Mode.STREAM,
|
||||
)
|
||||
),
|
||||
id="stream",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_redacted_thinking_is_sent_back_to_bedrock_on_the_next_turn(
|
||||
self, proxy: ProxyClient, resources: ResourceManager, stream: bool
|
||||
) -> None:
|
||||
edges: Final = _edges(resources)
|
||||
model: Final = _deploy(proxy, resources, edges, CONVERSE)
|
||||
key: Final = resources.key()
|
||||
send: Final = _chat_stream if stream else _chat
|
||||
question: Final = ChatMessage(role="user", content=REDACTED_THINKING_TRIGGER)
|
||||
|
||||
first: Final = send(proxy, key, ChatBody(model=model, messages=[question], thinking=THINKING, max_tokens=2048))
|
||||
redacted: Final = _redacted_data(first)
|
||||
|
||||
second: Final = send(proxy, key, _replay(model, question, first))
|
||||
|
||||
assert (second.content or "").strip(), f"Bedrock answered the replayed turn with no text: {second}"
|
||||
assert _converse_redacted(edges.capture.last_body("/converse")) == tuple(_b64(data) for data in redacted)
|
||||
|
||||
@pytest.mark.provider_edge_host
|
||||
@meta(
|
||||
Subject(
|
||||
domain=Domain.LLM_TRANSLATION,
|
||||
route=Route.CHAT_COMPLETIONS,
|
||||
providers=(Provider.BEDROCK,),
|
||||
models=(BACKEND,),
|
||||
capabilities=(Capability.FUNCTION_CALLING, Capability.REASONING),
|
||||
mode=Mode.NONSTREAM,
|
||||
)
|
||||
)
|
||||
def test_redacted_thinking_before_a_tool_call_is_sent_back_with_the_tool_result(
|
||||
self, proxy: ProxyClient, resources: ResourceManager
|
||||
) -> None:
|
||||
edges: Final = _edges(resources)
|
||||
model: Final = _deploy(proxy, resources, edges, CONVERSE)
|
||||
key: Final = resources.key()
|
||||
question: Final = ChatMessage(
|
||||
role="user", content=f"{REDACTED_THINKING_TRIGGER}\n\nWhat is the weather in Paris? Use the tool."
|
||||
)
|
||||
|
||||
first: Final = _chat(
|
||||
proxy,
|
||||
key,
|
||||
ChatBody(model=model, messages=[question], tools=[WEATHER_TOOL], thinking=THINKING, max_tokens=2048),
|
||||
)
|
||||
redacted: Final = _redacted_data(first)
|
||||
calls: Final = tuple(call for call in first.tool_calls or () if call.id)
|
||||
assert calls, f"turn 1 did not call get_weather: {first}"
|
||||
|
||||
second: Final = _chat(
|
||||
proxy,
|
||||
key,
|
||||
ChatBody(
|
||||
model=model,
|
||||
tools=[WEATHER_TOOL],
|
||||
thinking=THINKING,
|
||||
max_tokens=2048,
|
||||
messages=[
|
||||
question,
|
||||
ChatAssistantTurn(
|
||||
content=first.content, thinking_blocks=first.thinking_blocks, tool_calls=list(calls)
|
||||
),
|
||||
*(
|
||||
ChatToolResultTurn(tool_call_id=call.id or "", content="18 degrees C and sunny")
|
||||
for call in calls
|
||||
),
|
||||
],
|
||||
),
|
||||
)
|
||||
|
||||
assert (second.content or "").strip(), f"Bedrock answered the tool result with no text: {second}"
|
||||
assert _converse_redacted(edges.capture.last_body("/converse")) == tuple(_b64(data) for data in redacted)
|
||||
|
||||
@pytest.mark.provider_edge_host
|
||||
@pytest.mark.parametrize(
|
||||
"origin, destination",
|
||||
[
|
||||
_route_pair(origin, destination)
|
||||
for origin, destination in (
|
||||
("anthropic", "anthropic"),
|
||||
("anthropic", "invoke"),
|
||||
("anthropic", "converse"),
|
||||
("invoke", "anthropic"),
|
||||
("invoke", "invoke"),
|
||||
("invoke", "converse"),
|
||||
("converse", "anthropic"),
|
||||
("converse", "invoke"),
|
||||
)
|
||||
],
|
||||
)
|
||||
def test_redacted_thinking_minted_on_one_claude_route_replays_on_another(
|
||||
self, proxy: ProxyClient, resources: ResourceManager, origin: str, destination: str
|
||||
) -> None:
|
||||
"""A model group can mix the Anthropic API, Bedrock Invoke and Bedrock Converse, so a turn minted on one
|
||||
is replayed on another. Converse carries ``data`` base64-wrapped as redactedContent, the others carry it
|
||||
as is, and Claude 400s with "Invalid data in redacted_thinking block" on any other encoding"""
|
||||
edges: Final = _edges(resources)
|
||||
target: Final = CLAUDE_ROUTES[destination]
|
||||
origin_model: Final = _deploy(proxy, resources, edges, CLAUDE_ROUTES[origin])
|
||||
destination_model: Final = _deploy(proxy, resources, edges, target)
|
||||
key: Final = resources.key()
|
||||
question: Final = ChatMessage(role="user", content=REDACTED_THINKING_TRIGGER)
|
||||
|
||||
first: Final = _chat(
|
||||
proxy, key, ChatBody(model=origin_model, messages=[question], thinking=THINKING, max_tokens=2048)
|
||||
)
|
||||
redacted: Final = _redacted_data(first)
|
||||
|
||||
second: Final = _chat(proxy, key, _replay(destination_model, question, first))
|
||||
|
||||
assert (second.content or "").strip(), f"Claude answered the replayed turn with no text: {second}"
|
||||
assert target.replayed(edges.capture.last_body(target.path_marker)) == tuple(
|
||||
target.on_the_wire(data) for data in redacted
|
||||
)
|
||||
|
||||
@pytest.mark.provider_edge_host
|
||||
@meta(
|
||||
Subject(
|
||||
domain=Domain.LLM_TRANSLATION,
|
||||
route=Route.CHAT_COMPLETIONS,
|
||||
providers=(Provider.BEDROCK, Provider.OPENAI),
|
||||
models=(OPENAI_BACKEND, BACKEND),
|
||||
capabilities=(Capability.REASONING,),
|
||||
mode=Mode.NONSTREAM,
|
||||
)
|
||||
)
|
||||
def test_openai_encrypted_reasoning_from_messages_is_not_sent_to_claude_on_converse(
|
||||
self, proxy: ProxyClient, resources: ResourceManager
|
||||
) -> None:
|
||||
"""/v1/messages to an OpenAI model hands back its encrypted reasoning as thinking blocks carrying a
|
||||
LiteLLM blob only OpenAI can read. Claude rejects that blob, so replaying the turn to Claude leaves it out"""
|
||||
edges: Final = _edges(resources)
|
||||
gpt: Final = _create(
|
||||
proxy, resources, LiteLLMParamsBody(model=OPENAI_BACKEND, api_key="os.environ/OPENAI_API_KEY")
|
||||
)
|
||||
claude: Final = _deploy(proxy, resources, edges, CONVERSE)
|
||||
key: Final = resources.key()
|
||||
question: Final = ChatMessage(
|
||||
role="user",
|
||||
content="How many primes are there between 100 and 200? Think it through, then answer with just the number.",
|
||||
)
|
||||
|
||||
first: Final = unwrap(
|
||||
proxy.messages(
|
||||
key, AnthropicMessagesBody(model=gpt, messages=[question], max_tokens=8192, thinking=OPENAI_THINKING)
|
||||
)
|
||||
)
|
||||
blocks: Final = tuple(
|
||||
ThinkingBlock.model_validate(block.model_dump())
|
||||
for block in first.content or ()
|
||||
if block.type in ("thinking", "redacted_thinking")
|
||||
)
|
||||
encrypted: Final = tuple(
|
||||
block for block in blocks if (block.data or block.signature or "").startswith(ENCRYPTED_REASONING_PREFIX)
|
||||
)
|
||||
assert encrypted, f"turn 1 came back without encrypted reasoning: {first.content}"
|
||||
text: Final = "".join(block.text or "" for block in first.content or () if block.type == "text")
|
||||
|
||||
second: Final = _chat(
|
||||
proxy, key, _replay(claude, question, OutMessage(content=text, thinking_blocks=list(blocks)))
|
||||
)
|
||||
|
||||
assert (second.content or "").strip(), f"Claude answered the replayed turn with no text: {second}"
|
||||
assert _converse_reasoning(edges.capture.last_body("/converse")) == ()
|
||||
|
|
@ -11,9 +11,7 @@ from __future__ import annotations
|
|||
|
||||
import contextlib
|
||||
import json
|
||||
import threading
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass, field
|
||||
from collections.abc import Callable
|
||||
from types import MappingProxyType
|
||||
from typing import Final, cast
|
||||
|
||||
|
|
@ -39,7 +37,7 @@ from openai.types.responses import (
|
|||
ResponseReasoningItem,
|
||||
)
|
||||
from provider_edge import LiveEdge, start_provider_edge
|
||||
from provider_edge_bedrock import bedrock_signer
|
||||
from provider_edge_bedrock import ConverseBodyCapture, bedrock_signer
|
||||
from proxy_client import ProxyClient
|
||||
from pydantic import BaseModel, TypeAdapter
|
||||
from responses_helpers import AZURE_OPENAI_BACKEND, azure_openai_params
|
||||
|
|
@ -71,26 +69,6 @@ class ConverseRequestBody(BaseModel):
|
|||
additionalModelRequestFields: dict[str, str] | None = None
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ConverseRequestCapture:
|
||||
"""The Converse bodies the proxy actually sent upstream, as seen by a live
|
||||
edge sitting between the proxy and Bedrock."""
|
||||
|
||||
_bodies: list[ConverseRequestBody] = field(default_factory=list)
|
||||
_lock: threading.Lock = field(default_factory=threading.Lock)
|
||||
|
||||
def observe(self, url: str, headers: Mapping[str, str], body: bytes | None) -> None:
|
||||
if body is None or "/converse" not in url:
|
||||
return
|
||||
with self._lock:
|
||||
self._bodies.append(ConverseRequestBody.model_validate_json(body))
|
||||
|
||||
@property
|
||||
def bodies(self) -> tuple[ConverseRequestBody, ...]:
|
||||
with self._lock:
|
||||
return tuple(self._bodies)
|
||||
|
||||
|
||||
WEATHER_TOOL: FunctionToolParam = {
|
||||
"type": "function",
|
||||
"name": "get_weather",
|
||||
|
|
@ -919,7 +897,7 @@ class TestResponses:
|
|||
"""Judges the Converse bodies the edge captured, not the reply: Claude on
|
||||
Bedrock rejects the forwarded field with a 400, which the chat leg's
|
||||
``Result`` carries as a value and the OpenAI SDK raises."""
|
||||
capture: Final = ConverseRequestCapture()
|
||||
capture: Final = ConverseBodyCapture()
|
||||
edge: Final = start_provider_edge(
|
||||
LiveEdge(observe_request=capture.observe, sign=bedrock_signer(BEDROCK_EDGE_REGION)),
|
||||
mounts=MappingProxyType(
|
||||
|
|
@ -964,10 +942,11 @@ class TestResponses:
|
|||
),
|
||||
)
|
||||
|
||||
forwarded: Final = tuple(body.additionalModelRequestFields for body in capture.bodies)
|
||||
bodies: Final = tuple(ConverseRequestBody.model_validate_json(body) for body in capture.drain())
|
||||
forwarded: Final = tuple(body.additionalModelRequestFields for body in bodies)
|
||||
assert forwarded, f"{endpoint} produced no Bedrock Converse request"
|
||||
assert forwarded == ({"safety_identifier": safety_identifier},) * len(forwarded), (
|
||||
f"{endpoint} did not forward safety_identifier to Bedrock Converse on every attempt: {capture.bodies}"
|
||||
f"{endpoint} did not forward safety_identifier to Bedrock Converse on every attempt: {bodies}"
|
||||
)
|
||||
|
||||
@pytest.mark.skip(
|
||||
|
|
|
|||
|
|
@ -638,6 +638,7 @@ class AnthropicMessagesBody(BaseModel):
|
|||
stream: bool | None = None
|
||||
tools: list[AnthropicTool] | None = None
|
||||
tool_choice: AnthropicToolChoice | None = None
|
||||
thinking: ThinkingParam | None = None
|
||||
guardrails: list[str] | None = None
|
||||
cache: dict[str, bool] | None = {"no-cache": True}
|
||||
|
||||
|
|
|
|||
|
|
@ -22,8 +22,9 @@ request a permanent miss.
|
|||
from __future__ import annotations
|
||||
|
||||
import functools
|
||||
import queue
|
||||
from collections.abc import Callable, Mapping
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Final
|
||||
|
||||
from botocore.auth import SigV4Auth
|
||||
|
|
@ -70,3 +71,19 @@ def bedrock_signer(region: str, credentials: Callable[[], Credentials] = pod_cre
|
|||
"""Credentials are resolved on the first signed request, not here, so a run
|
||||
that mounts Bedrock but never calls it needs no AWS identity at all."""
|
||||
return BedrockSigner(region, credentials)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ConverseBodyCapture:
|
||||
"""The raw Converse bodies the proxy sent upstream, seen by a live edge
|
||||
between the proxy and Bedrock. Pass ``observe`` as the edge's
|
||||
``observe_request`` and ``drain`` once the calls under test are done."""
|
||||
|
||||
_bodies: queue.SimpleQueue[bytes] = field(default_factory=queue.SimpleQueue)
|
||||
|
||||
def observe(self, url: str, headers: Mapping[str, str], body: bytes | None) -> None:
|
||||
if body is not None and "/converse" in url:
|
||||
self._bodies.put(body)
|
||||
|
||||
def drain(self) -> tuple[bytes, ...]:
|
||||
return tuple(self._bodies.get_nowait() for _ in range(self._bodies.qsize()))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,415 @@
|
|||
import base64
|
||||
import json
|
||||
from collections.abc import Callable, Mapping
|
||||
from itertools import chain
|
||||
from typing import Final
|
||||
from urllib.parse import unquote
|
||||
|
||||
import pytest
|
||||
from integration._support.client import Gateway
|
||||
from integration._support.upstream import _aws_event_frame
|
||||
from integration._support.wire import Reply, Request, Wire, wire_server
|
||||
from pydantic import BaseModel, ConfigDict, JsonValue
|
||||
|
||||
CLAUDE_MODEL_ID: Final = "us.anthropic.claude-sonnet-4-6"
|
||||
NOVA_MODEL_ID: Final = "us.amazon.nova-pro-v1:0"
|
||||
APPLICATION_PROFILE_ARN: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/abc123"
|
||||
TOKEN: Final = "synthetic-bedrock-bearer"
|
||||
# Synthetic stand-in shaped like the `data` Bedrock Invoke returned for us.anthropic.claude-sonnet-4-6 in
|
||||
# us-west-2 on 2026-10-14: base64 of a protobuf message, so it opens with "E" + the 0x12 field tag. The
|
||||
# real value embeds the caller's AWS account id, so it is not committed.
|
||||
NATIVE_REDACTED: Final = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
CONVERSE_REDACTED: Final = base64.b64encode(NATIVE_REDACTED.encode()).decode()
|
||||
OPENAI_ENCRYPTED: Final = "litellm_encrypted_reasoning:gAAAAABsynthetic-openai-reasoning=="
|
||||
ANTHROPIC_MODEL: Final = "claude-sonnet-4-6"
|
||||
ANTHROPIC_REPLY: Final[dict[str, JsonValue]] = {
|
||||
"id": "msg_redacted_replay",
|
||||
"type": "message",
|
||||
"role": "assistant",
|
||||
"model": ANTHROPIC_MODEL,
|
||||
"content": [{"type": "text", "text": "redacted thinking replay control"}],
|
||||
"stop_reason": "end_turn",
|
||||
"stop_sequence": None,
|
||||
"usage": {"input_tokens": 11, "output_tokens": 4},
|
||||
}
|
||||
FIRST_ANSWER: Final = "I looked into it."
|
||||
FINAL_ANSWER: Final = "redacted thinking replay control"
|
||||
QUESTION: Final = "What's the weather in Paris?"
|
||||
FOLLOW_UP: Final = "Reply with the word ok."
|
||||
THINKING: Final[dict[str, JsonValue]] = {"type": "enabled", "budget_tokens": 1024}
|
||||
WEATHER_TOOL: Final[dict[str, JsonValue]] = {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
|
||||
},
|
||||
}
|
||||
TOOL_USE: Final[dict[str, JsonValue]] = {
|
||||
"toolUseId": "tooluse_weather_1",
|
||||
"name": "get_weather",
|
||||
"input": {"city": "Paris"},
|
||||
}
|
||||
USAGE: Final[dict[str, JsonValue]] = {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}
|
||||
EVENT_STREAM: Final = "application/vnd.amazon.eventstream"
|
||||
|
||||
|
||||
class _Frozen(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
|
||||
class _Message(_Frozen):
|
||||
content: str | None = None
|
||||
thinking_blocks: tuple[dict[str, JsonValue], ...] = ()
|
||||
tool_calls: tuple[dict[str, JsonValue], ...] = ()
|
||||
|
||||
|
||||
class _Choice(_Frozen):
|
||||
message: _Message
|
||||
|
||||
|
||||
class _Completion(_Frozen):
|
||||
choices: tuple[_Choice, ...]
|
||||
|
||||
|
||||
class _Delta(_Frozen):
|
||||
content: str | None = None
|
||||
thinking_blocks: tuple[dict[str, JsonValue], ...] | None = None
|
||||
|
||||
|
||||
class _StreamChoice(_Frozen):
|
||||
delta: _Delta
|
||||
|
||||
|
||||
class _Chunk(_Frozen):
|
||||
choices: tuple[_StreamChoice, ...] = ()
|
||||
|
||||
|
||||
class _ConverseMessage(_Frozen):
|
||||
role: str
|
||||
content: tuple[dict[str, JsonValue], ...]
|
||||
|
||||
|
||||
class _ConverseRequest(_Frozen):
|
||||
messages: tuple[_ConverseMessage, ...]
|
||||
|
||||
|
||||
class _AnthropicMessage(_Frozen):
|
||||
role: str
|
||||
content: str | tuple[dict[str, JsonValue], ...]
|
||||
|
||||
|
||||
class _AnthropicRequest(_Frozen):
|
||||
messages: tuple[_AnthropicMessage, ...]
|
||||
|
||||
|
||||
def _first_turn_content(with_tool: bool) -> list[JsonValue]:
|
||||
return [
|
||||
{"reasoningContent": {"redactedContent": CONVERSE_REDACTED}},
|
||||
{"text": FIRST_ANSWER},
|
||||
*([{"toolUse": TOOL_USE}] if with_tool else []),
|
||||
]
|
||||
|
||||
|
||||
def _converse_reply(content: list[JsonValue], stop_reason: str) -> Reply:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"output": {"message": {"role": "assistant", "content": content}},
|
||||
"stopReason": stop_reason,
|
||||
"usage": USAGE,
|
||||
"metrics": {"latencyMs": 1},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
def _frame(event_type: str, payload: Mapping[str, JsonValue]) -> bytes:
|
||||
return _aws_event_frame(event_type, payload, "sc", "u")
|
||||
|
||||
|
||||
def _stream_reply(*, first_turn: bool) -> Reply:
|
||||
reasoning: Final = (
|
||||
(
|
||||
_frame(
|
||||
"contentBlockDelta",
|
||||
{"delta": {"reasoningContent": {"redactedContent": CONVERSE_REDACTED}}, "contentBlockIndex": 0},
|
||||
),
|
||||
_frame("contentBlockStop", {"contentBlockIndex": 0}),
|
||||
)
|
||||
if first_turn
|
||||
else ()
|
||||
)
|
||||
text: Final = FIRST_ANSWER if first_turn else FINAL_ANSWER
|
||||
return Reply(
|
||||
content_type=EVENT_STREAM,
|
||||
body=b"".join(
|
||||
(
|
||||
_frame("messageStart", {"role": "assistant"}),
|
||||
*reasoning,
|
||||
_frame("contentBlockDelta", {"delta": {"text": text}, "contentBlockIndex": 1}),
|
||||
_frame("contentBlockStop", {"contentBlockIndex": 1}),
|
||||
_frame("messageStop", {"stopReason": "end_turn"}),
|
||||
_frame("metadata", {"usage": USAGE}),
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _is_replay(request: Request) -> bool:
|
||||
return any(message.role == "assistant" for message in _ConverseRequest.model_validate_json(request.body).messages)
|
||||
|
||||
|
||||
def _bedrock_peer(*, with_tool: bool = False) -> Callable[[Request], Reply]:
|
||||
"""Turn 1 answers with a redacted reasoning block, the way Bedrock does for a safety-redacted Claude turn.
|
||||
Turn 2, the one replaying that history, gets a plain answer."""
|
||||
|
||||
def respond(request: Request) -> Reply:
|
||||
if request.target == "/v1/messages":
|
||||
return Reply(body=json.dumps(ANTHROPIC_REPLY).encode())
|
||||
if unquote(request.target).endswith("/converse-stream"):
|
||||
return _stream_reply(first_turn=not _is_replay(request))
|
||||
if _is_replay(request):
|
||||
return _converse_reply([{"text": FINAL_ANSWER}], "end_turn")
|
||||
return _converse_reply(_first_turn_content(with_tool), "tool_use" if with_tool else "end_turn")
|
||||
|
||||
return respond
|
||||
|
||||
|
||||
def _chat(gateway: Gateway, body: Mapping[str, JsonValue]) -> _Message:
|
||||
response: Final = gateway.request("POST", "/v1/chat/completions", {**body, "cache": {"no-cache": True}})
|
||||
assert response.status_code == 200, response.text
|
||||
return _Completion.model_validate_json(response.content).choices[0].message
|
||||
|
||||
|
||||
def _chat_stream(gateway: Gateway, body: Mapping[str, JsonValue]) -> _Message:
|
||||
with gateway.client.stream(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
json={**body, "stream": True, "cache": {"no-cache": True}},
|
||||
headers={"Authorization": f"Bearer {gateway.key}"},
|
||||
) as response:
|
||||
assert response.status_code == 200, response.read()
|
||||
payloads: Final = tuple(
|
||||
line[6:] for line in response.iter_lines() if line.startswith("data: ") and line != "data: [DONE]"
|
||||
)
|
||||
chunks: Final = tuple(_Chunk.model_validate_json(payload) for payload in payloads)
|
||||
deltas: Final = tuple(choice.delta for choice in chain.from_iterable(chunk.choices for chunk in chunks))
|
||||
return _Message(
|
||||
content="".join(delta.content or "" for delta in deltas),
|
||||
thinking_blocks=tuple(chain.from_iterable(delta.thinking_blocks or () for delta in deltas)),
|
||||
)
|
||||
|
||||
|
||||
def _replayed_assistant_turn(wire: Wire) -> _ConverseMessage:
|
||||
received: Final = tuple(_ConverseRequest.model_validate_json(request.body) for request in wire.drain())
|
||||
return next(message for message in received[-1].messages if message.role == "assistant")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("stream", [False, True], ids=["nonstream", "stream"])
|
||||
def test_claude_redacted_thinking_round_trips_back_to_converse_as_redacted_content(
|
||||
gateway: Gateway, stream: bool
|
||||
) -> None:
|
||||
send: Final = _chat_stream if stream else _chat
|
||||
with wire_server(_bedrock_peer()) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(
|
||||
model=f"bedrock/converse/{CLAUDE_MODEL_ID}",
|
||||
api_key=TOKEN,
|
||||
aws_region_name="us-east-1",
|
||||
aws_bedrock_runtime_endpoint=wire.url,
|
||||
)
|
||||
question: Final[dict[str, JsonValue]] = {"role": "user", "content": QUESTION}
|
||||
first: Final = send(gateway, {"model": model, "messages": [question], "thinking": THINKING, "max_tokens": 2048})
|
||||
assert first.thinking_blocks == ({"type": "redacted_thinking", "data": NATIVE_REDACTED},)
|
||||
|
||||
second: Final = send(
|
||||
gateway,
|
||||
{
|
||||
"model": model,
|
||||
"thinking": THINKING,
|
||||
"max_tokens": 2048,
|
||||
"messages": [
|
||||
question,
|
||||
{"role": "assistant", "content": first.content, "thinking_blocks": list(first.thinking_blocks)},
|
||||
{"role": "user", "content": FOLLOW_UP},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
assert second.content == FINAL_ANSWER
|
||||
assert _replayed_assistant_turn(wire).content == (
|
||||
{"reasoningContent": {"redactedContent": CONVERSE_REDACTED}},
|
||||
{"text": FIRST_ANSWER},
|
||||
)
|
||||
|
||||
|
||||
def test_claude_redacted_thinking_before_a_tool_call_is_replayed_with_the_tool_result(gateway: Gateway) -> None:
|
||||
with wire_server(_bedrock_peer(with_tool=True)) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(
|
||||
model=f"bedrock/converse/{CLAUDE_MODEL_ID}",
|
||||
api_key=TOKEN,
|
||||
aws_region_name="us-east-1",
|
||||
aws_bedrock_runtime_endpoint=wire.url,
|
||||
)
|
||||
question: Final[dict[str, JsonValue]] = {"role": "user", "content": QUESTION}
|
||||
first: Final = _chat(
|
||||
gateway,
|
||||
{"model": model, "messages": [question], "tools": [WEATHER_TOOL], "thinking": THINKING, "max_tokens": 2048},
|
||||
)
|
||||
assert first.thinking_blocks == ({"type": "redacted_thinking", "data": NATIVE_REDACTED},)
|
||||
assert len(first.tool_calls) == 1
|
||||
|
||||
second: Final = _chat(
|
||||
gateway,
|
||||
{
|
||||
"model": model,
|
||||
"tools": [WEATHER_TOOL],
|
||||
"thinking": THINKING,
|
||||
"max_tokens": 2048,
|
||||
"messages": [
|
||||
question,
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": first.content,
|
||||
"thinking_blocks": list(first.thinking_blocks),
|
||||
"tool_calls": list(first.tool_calls),
|
||||
},
|
||||
{"role": "tool", "tool_call_id": str(TOOL_USE["toolUseId"]), "content": "18C, sunny"},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
assert second.content == FINAL_ANSWER
|
||||
assert _replayed_assistant_turn(wire).content == (
|
||||
{"reasoningContent": {"redactedContent": CONVERSE_REDACTED}},
|
||||
{"text": FIRST_ANSWER},
|
||||
{"toolUse": TOOL_USE},
|
||||
)
|
||||
|
||||
|
||||
_CLAUDE: Final[dict[str, JsonValue]] = {"model": f"bedrock/converse/{CLAUDE_MODEL_ID}"}
|
||||
|
||||
|
||||
def _redacted(data: str) -> dict[str, JsonValue]:
|
||||
return {"type": "redacted_thinking", "data": data}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"deployment, history_block, sent_data",
|
||||
[
|
||||
pytest.param({"model": f"bedrock/converse/{NOVA_MODEL_ID}"}, _redacted(NATIVE_REDACTED), None, id="nova"),
|
||||
pytest.param(
|
||||
{"model": f"bedrock/converse/{APPLICATION_PROFILE_ARN}"},
|
||||
_redacted(NATIVE_REDACTED),
|
||||
None,
|
||||
id="application-profile-arn-without-base-model",
|
||||
),
|
||||
pytest.param(
|
||||
{
|
||||
"model": f"bedrock/converse/{APPLICATION_PROFILE_ARN}",
|
||||
"model_info": {"base_model": f"bedrock/{CLAUDE_MODEL_ID}"},
|
||||
},
|
||||
_redacted(NATIVE_REDACTED),
|
||||
CONVERSE_REDACTED,
|
||||
id="application-profile-arn-with-claude-base-model",
|
||||
),
|
||||
pytest.param(
|
||||
{**_CLAUDE, "model_id": APPLICATION_PROFILE_ARN},
|
||||
_redacted(NATIVE_REDACTED),
|
||||
CONVERSE_REDACTED,
|
||||
id="claude-model-id-override",
|
||||
),
|
||||
pytest.param(_CLAUDE, _redacted(NATIVE_REDACTED), CONVERSE_REDACTED, id="claude-native-history"),
|
||||
pytest.param(
|
||||
_CLAUDE, _redacted(CONVERSE_REDACTED), CONVERSE_REDACTED, id="claude-history-saved-with-converse-wrapping"
|
||||
),
|
||||
pytest.param(_CLAUDE, _redacted(OPENAI_ENCRYPTED), None, id="claude-redacted-openai-encrypted-reasoning"),
|
||||
pytest.param(
|
||||
_CLAUDE,
|
||||
{"type": "thinking", "thinking": "Checking the forecast.", "signature": OPENAI_ENCRYPTED},
|
||||
None,
|
||||
id="claude-thinking-signed-with-openai-encrypted-reasoning",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_redacted_history_replayed_to_a_converse_target_carries_redacted_content_only_when_it_is_claude(
|
||||
gateway: Gateway, deployment: dict[str, JsonValue], history_block: dict[str, JsonValue], sent_data: str | None
|
||||
) -> None:
|
||||
"""A Router fallback or model switch replays a redacted history to a different Converse target. Non-Claude
|
||||
targets 400 on redactedContent, so only a target known to be Claude gets the block, always as base64 of
|
||||
Claude's own data. Claude cannot read OpenAI encrypted reasoning, so that block never reaches it"""
|
||||
model_info: Final = deployment.get("model_info")
|
||||
parameters: Final = {key: value for key, value in deployment.items() if key != "model_info"}
|
||||
with wire_server(_bedrock_peer()) as wire, gateway.scenario() as scenario:
|
||||
model: Final = scenario.model(
|
||||
model_info=model_info if isinstance(model_info, dict) else None,
|
||||
api_key=TOKEN,
|
||||
aws_region_name="us-east-1",
|
||||
aws_bedrock_runtime_endpoint=wire.url,
|
||||
**parameters,
|
||||
)
|
||||
|
||||
reply: Final = _chat(
|
||||
gateway,
|
||||
{
|
||||
"model": model,
|
||||
"max_tokens": 256,
|
||||
"messages": [
|
||||
{"role": "user", "content": QUESTION},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": FIRST_ANSWER,
|
||||
"thinking_blocks": [history_block],
|
||||
},
|
||||
{"role": "user", "content": FOLLOW_UP},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
assert reply.content == FINAL_ANSWER
|
||||
redacted: Final = () if sent_data is None else ({"reasoningContent": {"redactedContent": sent_data}},)
|
||||
assert _replayed_assistant_turn(wire).content == (*redacted, {"text": FIRST_ANSWER})
|
||||
|
||||
|
||||
@pytest.mark.parametrize("legacy_history", [False, True], ids=["converse-reply", "history-saved-with-converse-wrapping"])
|
||||
def test_converse_redacted_thinking_replays_to_the_anthropic_api_as_claude_native_data(
|
||||
gateway: Gateway, legacy_history: bool
|
||||
) -> None:
|
||||
"""A fallback from Claude on Converse to the Anthropic API (or Bedrock Invoke, which builds the same
|
||||
messages) must hand Claude its own data back, or Anthropic 400s with "Invalid `data` in `redacted_thinking`"
|
||||
"""
|
||||
with wire_server(_bedrock_peer()) as wire, gateway.scenario() as scenario:
|
||||
converse: Final = scenario.model(
|
||||
model=f"bedrock/converse/{CLAUDE_MODEL_ID}",
|
||||
api_key=TOKEN,
|
||||
aws_region_name="us-east-1",
|
||||
aws_bedrock_runtime_endpoint=wire.url,
|
||||
)
|
||||
anthropic: Final = scenario.model(
|
||||
model=f"anthropic/{ANTHROPIC_MODEL}", api_base=wire.url, api_key="synthetic-anthropic-key"
|
||||
)
|
||||
question: Final[dict[str, JsonValue]] = {"role": "user", "content": QUESTION}
|
||||
first: Final = _chat(
|
||||
gateway, {"model": converse, "messages": [question], "thinking": THINKING, "max_tokens": 2048}
|
||||
)
|
||||
history: Final = (_redacted(CONVERSE_REDACTED),) if legacy_history else first.thinking_blocks
|
||||
|
||||
second: Final = _chat(
|
||||
gateway,
|
||||
{
|
||||
"model": anthropic,
|
||||
"thinking": THINKING,
|
||||
"max_tokens": 2048,
|
||||
"messages": [
|
||||
question,
|
||||
{"role": "assistant", "content": first.content, "thinking_blocks": list(history)},
|
||||
{"role": "user", "content": FOLLOW_UP},
|
||||
],
|
||||
},
|
||||
)
|
||||
|
||||
assert second.content == FINAL_ANSWER
|
||||
sent: Final = _AnthropicRequest.model_validate_json(wire.drain()[-1].body)
|
||||
assert sent.messages[1].content == (
|
||||
{"type": "redacted_thinking", "data": NATIVE_REDACTED},
|
||||
{"type": "text", "text": FIRST_ANSWER},
|
||||
)
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
import base64
|
||||
import json
|
||||
from typing import Final
|
||||
|
||||
import pytest
|
||||
from integration._support.wire import Reply, Request, wire_server
|
||||
from pydantic import BaseModel, ConfigDict, JsonValue
|
||||
|
||||
from litellm import Router
|
||||
|
||||
APPLICATION_PROFILE_ARN: Final = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/abc123"
|
||||
REDACTED: Final = base64.b64encode(b"EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP").decode()
|
||||
FIRST_ANSWER: Final = "I looked into it."
|
||||
FINAL_ANSWER: Final = "redacted thinking router control"
|
||||
|
||||
|
||||
class _ConverseMessage(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
role: str
|
||||
content: tuple[dict[str, JsonValue], ...]
|
||||
|
||||
|
||||
class _ConverseRequest(BaseModel):
|
||||
model_config = ConfigDict(frozen=True)
|
||||
|
||||
messages: tuple[_ConverseMessage, ...]
|
||||
|
||||
|
||||
def _respond(_: Request) -> Reply:
|
||||
return Reply(
|
||||
body=json.dumps(
|
||||
{
|
||||
"output": {"message": {"role": "assistant", "content": [{"text": FINAL_ANSWER}]}},
|
||||
"stopReason": "end_turn",
|
||||
"usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15},
|
||||
"metrics": {"latencyMs": 1},
|
||||
}
|
||||
).encode()
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"model_info, replays_redacted",
|
||||
[
|
||||
pytest.param({"base_model": "bedrock/us.anthropic.claude-sonnet-4-6"}, True, id="claude-base-model"),
|
||||
pytest.param({}, False, id="no-base-model"),
|
||||
],
|
||||
)
|
||||
def test_sync_router_reads_base_model_to_replay_redacted_thinking_to_an_application_profile(
|
||||
model_info: dict[str, JsonValue], replays_redacted: bool
|
||||
) -> None:
|
||||
"""An application inference profile ARN hides its model, so the Router deployment's model_info.base_model is
|
||||
the only thing that tells the sync SDK path the target is Claude and can take redactedContent back"""
|
||||
with wire_server(_respond) as wire:
|
||||
router: Final = Router(
|
||||
model_list=[
|
||||
{
|
||||
"model_name": "claude-profile",
|
||||
"litellm_params": {
|
||||
"model": f"bedrock/converse/{APPLICATION_PROFILE_ARN}",
|
||||
"api_base": wire.url,
|
||||
"api_key": "synthetic-bedrock-bearer",
|
||||
"aws_region_name": "us-east-1",
|
||||
},
|
||||
"model_info": model_info,
|
||||
}
|
||||
],
|
||||
num_retries=0,
|
||||
)
|
||||
|
||||
response: Final = router.completion(
|
||||
model="claude-profile",
|
||||
max_tokens=256,
|
||||
messages=[
|
||||
{"role": "user", "content": "What's the weather in Paris?"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": FIRST_ANSWER,
|
||||
"thinking_blocks": [{"type": "redacted_thinking", "data": REDACTED}],
|
||||
},
|
||||
{"role": "user", "content": "Reply with the word ok."},
|
||||
],
|
||||
)
|
||||
|
||||
assert response.choices[0].message.content == FINAL_ANSWER
|
||||
received: Final = tuple(_ConverseRequest.model_validate_json(request.body) for request in wire.drain())
|
||||
assistant: Final = next(message for message in received[-1].messages if message.role == "assistant")
|
||||
redacted: Final = ({"reasoningContent": {"redactedContent": REDACTED}},) if replays_redacted else ()
|
||||
assert assistant.content == (*redacted, {"text": FIRST_ANSWER})
|
||||
|
|
@ -1,3 +1,4 @@
|
|||
import base64
|
||||
import copy
|
||||
import functools
|
||||
import json
|
||||
|
|
@ -13,6 +14,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
|||
TOOL_RESULT_IMAGE_PLACEHOLDER,
|
||||
add_system_prompt_to_messages,
|
||||
cache_control_as_prompt_cache_breakpoint,
|
||||
converse_redacted_content,
|
||||
encrypted_content_from_signature,
|
||||
encrypted_reasoning_signature,
|
||||
get_file_ids_from_messages,
|
||||
|
|
@ -23,6 +25,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
|||
hoist_images_from_tool_messages,
|
||||
is_encrypted_reasoning_block,
|
||||
merge_consecutive_system_messages,
|
||||
native_redacted_thinking_data,
|
||||
parse_tool_call_arguments,
|
||||
responses_reasoning_items_from_thinking_blocks,
|
||||
salvage_concatenated_tool_arguments,
|
||||
|
|
@ -2306,3 +2309,28 @@ def test_a_breakpoint_the_client_wrote_survives_a_per_part_cache_control() -> No
|
|||
"text": "rules",
|
||||
"prompt_cache_breakpoint": {"mode": "explicit", "ttl": "30m"},
|
||||
}
|
||||
|
||||
|
||||
_NATIVE_REDACTED_DATA: Final = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
_CONVERSE_REDACTED_DATA: Final = base64.b64encode(_NATIVE_REDACTED_DATA.encode()).decode()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"data, native",
|
||||
[
|
||||
pytest.param(_NATIVE_REDACTED_DATA, _NATIVE_REDACTED_DATA, id="claude-native"),
|
||||
pytest.param(_CONVERSE_REDACTED_DATA, _NATIVE_REDACTED_DATA, id="converse-wrapped-legacy-history"),
|
||||
pytest.param(
|
||||
encrypted_reasoning_signature("gAAAA_1"), encrypted_reasoning_signature("gAAAA_1"), id="openai-encrypted"
|
||||
),
|
||||
pytest.param("", "", id="blank"),
|
||||
pytest.param(base64.b64encode(b"not base64!").decode(), base64.b64encode(b"not base64!").decode(), id="b64-once"),
|
||||
],
|
||||
)
|
||||
def test_native_redacted_thinking_data_only_unwraps_the_converse_base64_layer(data: str, native: str) -> None:
|
||||
assert native_redacted_thinking_data(data) == native
|
||||
|
||||
|
||||
@pytest.mark.parametrize("data", [_NATIVE_REDACTED_DATA, _CONVERSE_REDACTED_DATA], ids=["native", "legacy-wrapped"])
|
||||
def test_converse_redacted_content_wraps_claude_native_data_exactly_once(data: str) -> None:
|
||||
assert converse_redacted_content(data) == _CONVERSE_REDACTED_DATA
|
||||
|
|
|
|||
|
|
@ -57,8 +57,12 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
|
|||
llama_2_chat_pt,
|
||||
prompt_factory,
|
||||
)
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import get_completion_messages
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
encrypted_reasoning_signature,
|
||||
get_completion_messages,
|
||||
)
|
||||
from litellm.llms.vertex_ai.gemini.transformation import gemini_convert_messages_with_history
|
||||
from litellm.types.llms.bedrock import BedrockConverseReasoningContentBlock
|
||||
from litellm.types.llms.openai import AllMessageValues
|
||||
|
||||
|
||||
|
|
@ -411,6 +415,175 @@ def test_bedrock_converse_assistant_with_empty_thinking_block_and_tool_calls():
|
|||
assert len(tool_use_blocks) == 2
|
||||
|
||||
|
||||
_NATIVE_REDACTED_DATA: Final = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
_CONVERSE_REDACTED_DATA: Final = base64.b64encode(_NATIVE_REDACTED_DATA.encode()).decode()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
@pytest.mark.parametrize("placement", ["thinking_blocks", "content"])
|
||||
async def test_bedrock_converse_replays_redacted_thinking_as_redacted_content(use_async: bool, placement: str):
|
||||
"""A redacted_thinking block that Bedrock returned must go back as reasoningContent.redactedContent,
|
||||
in order, ahead of text and toolUse. It used to collapse into an empty unsigned reasoningText and get dropped."""
|
||||
bedrock_reasoning_blocks: Final[list[BedrockConverseReasoningContentBlock]] = [
|
||||
{"reasoningText": {"text": "Let me look this up.", "signature": "sig-abc"}},
|
||||
{"redactedContent": _CONVERSE_REDACTED_DATA},
|
||||
]
|
||||
thinking_blocks: Final = litellm.AmazonConverseConfig()._transform_thinking_blocks(bedrock_reasoning_blocks)
|
||||
assistant_text: Final = "Checking the weather."
|
||||
assistant_message: Final = (
|
||||
{"role": "assistant", "content": assistant_text, "thinking_blocks": thinking_blocks}
|
||||
if placement == "thinking_blocks"
|
||||
else {"role": "assistant", "content": [*thinking_blocks, {"type": "text", "text": assistant_text}]}
|
||||
)
|
||||
messages: Final = [
|
||||
{"role": "user", "content": "What's the weather in Paris?"},
|
||||
{
|
||||
**assistant_message,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "tooluse_weather_1",
|
||||
"type": "function",
|
||||
"function": {"name": "get_weather", "arguments": '{"city": "Paris"}'},
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "tooluse_weather_1", "content": "18C, sunny"},
|
||||
]
|
||||
model: Final = "us.anthropic.claude-sonnet-4-6"
|
||||
|
||||
result: Final = (
|
||||
await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async(
|
||||
messages=messages, model=model, llm_provider="bedrock"
|
||||
)
|
||||
if use_async
|
||||
else _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock")
|
||||
)
|
||||
|
||||
assistant_content: Final = next(m["content"] for m in result if m["role"] == "assistant")
|
||||
assert assistant_content == [
|
||||
{"reasoningContent": bedrock_reasoning_blocks[0]},
|
||||
{"reasoningContent": bedrock_reasoning_blocks[1]},
|
||||
{"text": assistant_text},
|
||||
{"toolUse": {"toolUseId": "tooluse_weather_1", "name": "get_weather", "input": {"city": "Paris"}}},
|
||||
]
|
||||
|
||||
|
||||
_SENT_REDACTED: Final = [{"reasoningContent": {"redactedContent": _CONVERSE_REDACTED_DATA}}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
@pytest.mark.parametrize("placement", ["thinking_blocks", "content"])
|
||||
@pytest.mark.parametrize(
|
||||
"block, sent",
|
||||
[
|
||||
pytest.param({"type": "redacted_thinking", "data": _NATIVE_REDACTED_DATA}, _SENT_REDACTED, id="native"),
|
||||
pytest.param(
|
||||
{"type": "redacted_thinking", "data": _CONVERSE_REDACTED_DATA}, _SENT_REDACTED, id="legacy-wrapped"
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAAA_1")}, [], id="openai-redacted"
|
||||
),
|
||||
pytest.param(
|
||||
{"type": "thinking", "thinking": "Plan.", "signature": encrypted_reasoning_signature("gAAAA_1")},
|
||||
[],
|
||||
id="openai-thinking",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_bedrock_converse_replays_reasoning_to_claude_only_as_base64_of_claude_native_data(
|
||||
use_async: bool, placement: str, block: dict[str, str], sent: list[dict[str, dict[str, str]]]
|
||||
):
|
||||
"""Converse wants base64 of Claude's own ``data``, which the Anthropic API, Vertex, Bedrock Invoke and now
|
||||
Converse replies all hand out. Histories saved before Converse replies were decoded still hold it wrapped once.
|
||||
Claude cannot verify OpenAI encrypted reasoning from the Responses bridge, so Bedrock 400ed on it."""
|
||||
assistant_message: Final = (
|
||||
{"role": "assistant", "content": "hello", "thinking_blocks": [block]}
|
||||
if placement == "thinking_blocks"
|
||||
else {"role": "assistant", "content": [block, {"type": "text", "text": "hello"}]}
|
||||
)
|
||||
messages: Final = [{"role": "user", "content": "hi"}, assistant_message, {"role": "user", "content": "again"}]
|
||||
model: Final = "us.anthropic.claude-sonnet-4-6"
|
||||
|
||||
result: Final = (
|
||||
await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async(
|
||||
messages=messages, model=model, llm_provider="bedrock"
|
||||
)
|
||||
if use_async
|
||||
else _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock")
|
||||
)
|
||||
|
||||
assistant_content: Final = next(m["content"] for m in result if m["role"] == "assistant")
|
||||
assert assistant_content == [*sent, {"text": "hello"}]
|
||||
|
||||
|
||||
def test_translate_thinking_blocks_keeps_dropping_redacted_data_when_called_the_old_way():
|
||||
"""Outside callers that only pass the thinking blocks keep the old behavior, with no redactedContent sent."""
|
||||
assert BedrockConverseMessagesProcessor.translate_thinking_blocks_to_reasoning_content_blocks(
|
||||
[{"type": "redacted_thinking", "data": _CONVERSE_REDACTED_DATA}]
|
||||
) == [{"reasoningContent": {"reasoningText": {"text": ""}}}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("data", ["", " "])
|
||||
def test_bedrock_converse_drops_redacted_thinking_with_blank_data(data: str):
|
||||
"""Bedrock 400s on a blank redactedContent, so a redacted block with no payload is dropped, not sent."""
|
||||
messages: Final = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [{"type": "redacted_thinking", "data": data}, {"type": "text", "text": "hello"}],
|
||||
"thinking_blocks": [{"type": "redacted_thinking", "data": data}],
|
||||
},
|
||||
{"role": "user", "content": "again"},
|
||||
]
|
||||
|
||||
result: Final = _bedrock_converse_messages_pt(
|
||||
messages=messages, model="us.anthropic.claude-sonnet-4-6", llm_provider="bedrock"
|
||||
)
|
||||
|
||||
assistant_content: Final = next(m["content"] for m in result if m["role"] == "assistant")
|
||||
assert assistant_content == [{"text": "hello"}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
@pytest.mark.parametrize("placement", ["thinking_blocks", "content"])
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
[
|
||||
"us.amazon.nova-pro-v1:0",
|
||||
"us.meta.llama4-maverick-17b-instruct-v1:0",
|
||||
"us.deepseek.r1-v1:0",
|
||||
"openai.gpt-oss-120b-1:0",
|
||||
"arn:aws:bedrock:us-west-2:123456789012:application-inference-profile/abc123",
|
||||
],
|
||||
)
|
||||
async def test_bedrock_converse_keeps_dropping_redacted_thinking_for_non_claude_targets(
|
||||
use_async: bool, placement: str, model: str
|
||||
):
|
||||
"""Only Claude takes redactedContent back. Any other Converse target 400s on it, so a Claude history replayed
|
||||
there (a Router fallback or a model switch) drops the redacted block and keeps the call working."""
|
||||
redacted: Final = {"type": "redacted_thinking", "data": _CONVERSE_REDACTED_DATA}
|
||||
assistant_message: Final = (
|
||||
{"role": "assistant", "content": "hello", "thinking_blocks": [redacted]}
|
||||
if placement == "thinking_blocks"
|
||||
else {"role": "assistant", "content": [redacted, {"type": "text", "text": "hello"}]}
|
||||
)
|
||||
messages: Final = [{"role": "user", "content": "hi"}, assistant_message, {"role": "user", "content": "again"}]
|
||||
|
||||
result: Final = (
|
||||
await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async(
|
||||
messages=messages, model=model, llm_provider="bedrock"
|
||||
)
|
||||
if use_async
|
||||
else _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock")
|
||||
)
|
||||
|
||||
assistant_content: Final = next(m["content"] for m in result if m["role"] == "assistant")
|
||||
assert assistant_content == [{"text": "hello"}]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"thinking_block",
|
||||
[
|
||||
|
|
@ -458,6 +631,31 @@ def test_anthropic_messages_pt_drops_unsignable_thinking_block(thinking_block):
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("data", [_NATIVE_REDACTED_DATA, _CONVERSE_REDACTED_DATA], ids=["native", "legacy-wrapped"])
|
||||
def test_anthropic_messages_pt_replays_redacted_thinking_as_claude_native_data(data: str):
|
||||
"""Histories saved from Converse before its replies were decoded hold the data base64-wrapped, and the
|
||||
Anthropic API, Vertex and Bedrock Invoke 400 on it with "Invalid `data` in `redacted_thinking` block"."""
|
||||
messages: Final = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "hello",
|
||||
"thinking_blocks": [{"type": "redacted_thinking", "data": data, "cache_control": {"type": "ephemeral"}}],
|
||||
},
|
||||
{"role": "user", "content": "again"},
|
||||
]
|
||||
|
||||
result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic")
|
||||
|
||||
assert result[1] == {
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "redacted_thinking", "data": _NATIVE_REDACTED_DATA, "cache_control": {"type": "ephemeral"}},
|
||||
{"type": "text", "text": "hello"},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def test_anthropic_messages_pt_keeps_signed_thinking_block():
|
||||
"""A genuine Anthropic round-trip still holds its original signature, so that
|
||||
thinking block must be forwarded unchanged (we only drop unsignable blocks).
|
||||
|
|
@ -4346,7 +4544,7 @@ def test_is_unsignable_thinking_block_treats_whitespace_only_as_empty():
|
|||
the same as a fully empty string and drop the block.
|
||||
|
||||
The check lives in the shared `is_unsignable_thinking_block` helper, which
|
||||
`_drop_unsignable_thinking_blocks` calls standalone, so the whitespace-aware
|
||||
`_replayable_thinking_blocks` calls standalone, so the whitespace-aware
|
||||
test has to hold there rather than only at the factory call site.
|
||||
"""
|
||||
from litellm.litellm_core_utils.prompt_templates.common_utils import (
|
||||
|
|
|
|||
|
|
@ -1570,7 +1570,7 @@ class TestTranslateResponse:
|
|||
def test_thinking_blocks_are_dropped_when_replayed_to_anthropic(self, summaries, encrypted_content):
|
||||
"""Replaying this turn to an Anthropic model must not send a signature it cannot verify."""
|
||||
from litellm.litellm_core_utils.prompt_templates.factory import (
|
||||
_drop_unsignable_thinking_blocks,
|
||||
_replayable_thinking_blocks,
|
||||
)
|
||||
|
||||
response = _make_mock_response(
|
||||
|
|
@ -1578,7 +1578,7 @@ class TestTranslateResponse:
|
|||
)
|
||||
result: Any = _ADAPTER.translate_response(response)
|
||||
assert len(result["content"]) == 1
|
||||
assert _drop_unsignable_thinking_blocks(result["content"]) == []
|
||||
assert _replayable_thinking_blocks(result["content"]) == []
|
||||
|
||||
def test_usage_mapped_correctly(self):
|
||||
"""Input/output tokens from ResponseAPIUsage are mapped to AnthropicUsage."""
|
||||
|
|
|
|||
|
|
@ -1667,6 +1667,37 @@ class TestAnthropicThinkingSignatureSelfHeal:
|
|||
assert len(msgs[1]["content"]) == 2
|
||||
assert len(msgs[2]["content"]) == 4
|
||||
|
||||
def test_strip_encrypted_reasoning_unwraps_converse_wrapped_redacted_data(self):
|
||||
"""/v1/messages turns served by Bedrock Converse used to hand out the data base64-wrapped, which the
|
||||
Anthropic API and Bedrock Invoke reject with "Invalid `data` in `redacted_thinking` block"."""
|
||||
import base64
|
||||
|
||||
from litellm.llms.anthropic.common_utils import (
|
||||
strip_encrypted_reasoning_blocks_from_anthropic_messages,
|
||||
)
|
||||
|
||||
native = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
untouched = {"role": "assistant", "content": [{"type": "redacted_thinking", "data": native}]}
|
||||
msgs = [
|
||||
{"role": "user", "content": "Solve it."},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "redacted_thinking", "data": base64.b64encode(native.encode()).decode()},
|
||||
{"type": "text", "text": "The answer."},
|
||||
],
|
||||
},
|
||||
untouched,
|
||||
]
|
||||
|
||||
out = strip_encrypted_reasoning_blocks_from_anthropic_messages(msgs)
|
||||
|
||||
assert out[1] == {
|
||||
"role": "assistant",
|
||||
"content": [{"type": "redacted_thinking", "data": native}, {"type": "text", "text": "The answer."}],
|
||||
}
|
||||
assert out[2] is untouched
|
||||
|
||||
def test_strip_encrypted_reasoning_leaves_malformed_messages_for_the_provider_to_reject(self):
|
||||
"""A bare string in messages must reach Anthropic as a 400, not die in the stripper as a 500."""
|
||||
from litellm.llms.anthropic.common_utils import (
|
||||
|
|
|
|||
|
|
@ -618,23 +618,21 @@ def test_bedrock_converse_like_proxy_uses_bearer_api_key(respx_mock: respx.Route
|
|||
assert response.choices[0].message.content == "ok"
|
||||
|
||||
|
||||
def test_transform_thinking_blocks_with_redacted_content():
|
||||
thinking_blocks = [
|
||||
{
|
||||
"reasoningText": {
|
||||
"text": "This is a test",
|
||||
"signature": "test_signature",
|
||||
}
|
||||
},
|
||||
{
|
||||
"redactedContent": "This is a redacted content",
|
||||
},
|
||||
def test_transform_thinking_blocks_unwraps_redacted_content_to_claude_native_data():
|
||||
"""Converse returns Claude's redacted data base64-wrapped. Handing it out as is made the Anthropic API and
|
||||
Bedrock Invoke 400 with "Invalid `data` in `redacted_thinking` block" when the history was replayed there."""
|
||||
native_data: Final = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
thinking_blocks: Final = [
|
||||
{"reasoningText": {"text": "This is a test", "signature": "test_signature"}},
|
||||
{"redactedContent": base64.b64encode(native_data.encode()).decode()},
|
||||
]
|
||||
|
||||
transformed_thinking_blocks: Final = AmazonConverseConfig()._transform_thinking_blocks(thinking_blocks)
|
||||
|
||||
assert transformed_thinking_blocks == [
|
||||
{"type": "thinking", "thinking": "This is a test", "signature": "test_signature"},
|
||||
{"type": "redacted_thinking", "data": native_data},
|
||||
]
|
||||
config = AmazonConverseConfig()
|
||||
transformed_thinking_blocks = config._transform_thinking_blocks(thinking_blocks)
|
||||
assert len(transformed_thinking_blocks) == 2
|
||||
assert transformed_thinking_blocks[0]["type"] == "thinking"
|
||||
assert transformed_thinking_blocks[1]["type"] == "redacted_thinking"
|
||||
|
||||
|
||||
def test_apply_tool_call_transformation_if_needed():
|
||||
|
|
@ -1088,6 +1086,63 @@ def test_transform_request_resolves_an_inference_profile_through_its_base_model(
|
|||
assert properties == _LOOKAROUND_FREE_PROPERTIES
|
||||
|
||||
|
||||
_APPLICATION_PROFILE_ARN: Final = "arn:aws:bedrock:us-west-2:123456789012:application-inference-profile/abc123"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
@pytest.mark.parametrize(
|
||||
"model, optional_params, litellm_params, replays_redacted",
|
||||
[
|
||||
("us.anthropic.claude-sonnet-4-6", {}, {}, True),
|
||||
("arn:aws:bedrock:us-west-2:123456789012:inference-profile/us.anthropic.claude-sonnet-4-6", {}, {}, True),
|
||||
("anthropic.claude-sonnet-4-6", {"model_id": _APPLICATION_PROFILE_ARN}, {}, True),
|
||||
(_APPLICATION_PROFILE_ARN, {}, {"base_model": "anthropic.claude-sonnet-4-6"}, True),
|
||||
(_APPLICATION_PROFILE_ARN, {}, {"base_model": "bedrock/us.anthropic.claude-sonnet-4-6"}, True),
|
||||
(_APPLICATION_PROFILE_ARN, {}, {"base_model": "claude-sonnet-4-6"}, True),
|
||||
(_APPLICATION_PROFILE_ARN, {}, {}, False),
|
||||
(_APPLICATION_PROFILE_ARN, {}, {"base_model": "us.amazon.nova-pro-v1:0"}, False),
|
||||
("us.amazon.nova-pro-v1:0", {}, {}, False),
|
||||
("us.amazon.nova-pro-v1:0", {}, {"base_model": "anthropic.claude-sonnet-4-6"}, False),
|
||||
],
|
||||
)
|
||||
async def test_transform_request_replays_redacted_thinking_only_to_claude(
|
||||
use_async: bool,
|
||||
model: str,
|
||||
optional_params: dict[str, str],
|
||||
litellm_params: dict[str, str],
|
||||
replays_redacted: bool,
|
||||
):
|
||||
"""A model id that names its model decides on its own. An application inference profile ARN hides its model,
|
||||
so the deployment's ``base_model`` is what marks it as Claude, and without one the redacted block is dropped
|
||||
the way it is for every non-Claude target."""
|
||||
redacted_data: Final = base64.b64encode(b"EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP").decode()
|
||||
messages: Final = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": "hello",
|
||||
"thinking_blocks": [{"type": "redacted_thinking", "data": redacted_data}],
|
||||
},
|
||||
{"role": "user", "content": "again"},
|
||||
]
|
||||
config: Final = AmazonConverseConfig()
|
||||
kwargs: Final = dict(
|
||||
model=model, messages=messages, optional_params=dict(optional_params), litellm_params=dict(litellm_params)
|
||||
)
|
||||
|
||||
request: Final = (
|
||||
await config._async_transform_request(**kwargs) if use_async else config._transform_request(**kwargs)
|
||||
)
|
||||
|
||||
assert request["messages"][1] == {
|
||||
"role": "assistant",
|
||||
"content": [{"reasoningContent": {"redactedContent": redacted_data}}, {"text": "hello"}]
|
||||
if replays_redacted
|
||||
else [{"text": "hello"}],
|
||||
}
|
||||
|
||||
|
||||
def test_transform_request_drops_lookaround_regex_around_pre_formatted_tool_blocks():
|
||||
"""Blocks that arrive already in Bedrock shape, like Nova's grounding ``systemTool``, pass through as sent."""
|
||||
grounding: Final = {"systemTool": {"name": "nova_grounding"}}
|
||||
|
|
|
|||
|
|
@ -41,13 +41,17 @@ def bedrock_transformer() -> AmazonInvokeConfig:
|
|||
return AmazonInvokeConfig()
|
||||
|
||||
|
||||
def test_transform_thinking_blocks_with_redacted_content():
|
||||
thinking_block = {"redactedContent": "This is a redacted content"}
|
||||
decoder = AWSEventStreamDecoder(model="test")
|
||||
transformed_thinking_blocks = decoder.translate_thinking_blocks(thinking_block)
|
||||
assert len(transformed_thinking_blocks) == 1
|
||||
assert transformed_thinking_blocks[0]["type"] == "redacted_thinking"
|
||||
assert transformed_thinking_blocks[0]["data"] == "This is a redacted content"
|
||||
def test_transform_thinking_blocks_unwraps_converse_redacted_content_to_claude_native_data():
|
||||
"""Converse streams Claude's redacted data base64-wrapped. Handing it out as is made the Anthropic API and
|
||||
Bedrock Invoke 400 with "Invalid `data` in `redacted_thinking` block" when the history was replayed there."""
|
||||
native_data: Final = "EmwKAhgBEgy3va3pzix/LafPsn4aDFIT2Xlxh0L5L8rLVyIwxtE3rAFBa8cr3qpP"
|
||||
decoder: Final = AWSEventStreamDecoder(model="test")
|
||||
|
||||
thinking_blocks: Final = decoder.translate_thinking_blocks(
|
||||
{"redactedContent": base64.b64encode(native_data.encode()).decode()}
|
||||
)
|
||||
|
||||
assert thinking_blocks == [{"type": "redacted_thinking", "data": native_data}]
|
||||
|
||||
|
||||
def test_transform_tool_calls_index():
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue