fix(anthropic): unwrap tool-emulated json_schema envelopes on streams and tool-name wrappers

This commit is contained in:
mateo-berri 2026-10-11 11:16:43 -07:00
parent 3b843746c5
commit 53377d8f30
11 changed files with 1562 additions and 64 deletions

View file

@ -4,6 +4,7 @@ Calling + translation logic for anthropic's `/v1/messages` endpoint
import copy
import json
import re
from collections.abc import Callable, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Union, cast
@ -20,6 +21,7 @@ from litellm.anthropic_beta_headers_manager import (
from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
from litellm.litellm_core_utils.core_helpers import map_finish_reason
from litellm.litellm_core_utils.json_fragment_accumulator import JSONFragmentAccumulator
from litellm.llms.base_llm.base_utils import RESPONSE_FORMAT_ENVELOPE_NAME_KEYS, ResponseFormatSchema
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
HTTPHandler,
@ -57,7 +59,7 @@ from litellm.types.utils import (
from ...base import BaseLLM
from ..common_utils import AnthropicError, process_anthropic_headers
from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig
from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig, JsonModeRequest
if TYPE_CHECKING:
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
@ -69,6 +71,30 @@ def _loads_stream_chunk(payload: str) -> dict[str, object]:
return json.loads(payload)
_EnvelopeVerdict = Literal["undecided", "pass_through", "hold"]
_FIRST_OBJECT_KEY: Final = re.compile(r'^\s*\{\s*"((?:[^"\\]|\\.)*)"\s*:')
_OBJECT_KEY_PREFIX: Final = re.compile(r'^\s*(?:\{\s*(?:"(?:[^"\\]|\\.)*\\?(?:"\s*)?)?)?$')
_ENVELOPE_NAME_KEY_PATTERN: Final = "|".join(sorted(RESPONSE_FORMAT_ENVELOPE_NAME_KEYS))
_FIRST_NAME_VALUE: Final = re.compile(r'^\s*\{\s*"(?:' + _ENVELOPE_NAME_KEY_PATTERN + r')"\s*:\s*"((?:[^"\\]|\\.)*)"')
_NAME_VALUE_PREFIX: Final = re.compile(
r'^\s*\{\s*"(?:' + _ENVELOPE_NAME_KEY_PATTERN + r')"\s*:\s*(?:"(?:[^"\\]|\\.)*\\?)?$'
)
def _envelope_verdict(buffered_json: str, schema: ResponseFormatSchema, tool_name: str) -> _EnvelopeVerdict:
first_key: Final = _FIRST_OBJECT_KEY.match(buffered_json)
if first_key is None:
return "undecided" if _OBJECT_KEY_PREFIX.match(buffered_json) else "pass_through"
if first_key.group(1) not in schema.property_names:
return "hold"
if first_key.group(1) not in RESPONSE_FORMAT_ENVELOPE_NAME_KEYS:
return "pass_through"
first_value: Final = _FIRST_NAME_VALUE.match(buffered_json)
if first_value is None:
return "undecided" if _NAME_VALUE_PREFIX.match(buffered_json) else "pass_through"
return "hold" if first_value.group(1) == tool_name else "pass_through"
async def make_call(
client: AsyncHTTPHandler | None,
api_base: str,
@ -81,6 +107,7 @@ async def make_call(
json_mode: bool,
speed: str | None = None,
tool_name_reverse_map: dict[str, str] | None = None,
response_format_schema: ResponseFormatSchema | None = None,
) -> tuple["ModelResponseIterator", httpx.Headers]:
if client is None:
client = litellm.module_level_aclient
@ -116,6 +143,7 @@ async def make_call(
json_mode=json_mode,
speed=speed,
tool_name_reverse_map=tool_name_reverse_map,
response_format_schema=response_format_schema,
)
# LOGGING
@ -141,6 +169,7 @@ def make_sync_call(
json_mode: bool,
speed: str | None = None,
tool_name_reverse_map: dict[str, str] | None = None,
response_format_schema: ResponseFormatSchema | None = None,
) -> tuple["ModelResponseIterator", httpx.Headers]:
if client is None:
client = litellm.module_level_client # re-use a module level client
@ -184,6 +213,7 @@ def make_sync_call(
json_mode=json_mode,
speed=speed,
tool_name_reverse_map=tool_name_reverse_map,
response_format_schema=response_format_schema,
)
# LOGGING
@ -241,6 +271,9 @@ class AnthropicChatCompletion(BaseLLM):
tool_name_reverse_map=(
litellm_params.get(ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY) if isinstance(litellm_params, dict) else None
),
response_format_schema=(
JsonModeRequest.model_validate(optional_params).tool_schema() if json_mode and optional_params else None
),
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,
@ -485,6 +518,9 @@ class AnthropicChatCompletion(BaseLLM):
if isinstance(litellm_params, dict)
else None
),
response_format_schema=(
JsonModeRequest.model_validate(optional_params).tool_schema() if json_mode else None
),
)
return CustomStreamWrapper(
completion_stream=completion_stream,
@ -550,6 +586,7 @@ class ModelResponseIterator:
json_mode: bool | None = False,
speed: str | None = None,
tool_name_reverse_map: dict[str, str] | None = None,
response_format_schema: ResponseFormatSchema | None = None,
):
self.streaming_response = streaming_response
self.response_iterator = self.streaming_response
@ -573,6 +610,9 @@ class ModelResponseIterator:
self.is_response_format_tool: bool = False
# Track if we've converted any response_format tools (affects finish_reason)
self.converted_response_format_tool: bool = False
self.response_format_schema = response_format_schema
self._response_format_buffer: str = ""
self._response_format_verdict: _EnvelopeVerdict = "undecided"
# For handling partial JSON chunks from fragmentation
# See: https://github.com/BerriAI/litellm/issues/17473
@ -955,6 +995,8 @@ class ModelResponseIterator:
except (json.JSONDecodeError, TypeError):
pass
self._current_server_tool_id = None
if self.is_response_format_tool:
text = self._flush_response_format_buffer()
# Reset response_format tool tracking when block stops
self.is_response_format_tool = False
# Reset current content block type
@ -1072,6 +1114,10 @@ class ModelResponseIterator:
tool_name: Final = tool_use.get("function", {}).get("name", "")
self.is_response_format_tool = tool_name == RESPONSE_FORMAT_TOOL_NAME
if self.is_response_format_tool and self.response_format_schema is not None:
self.converted_response_format_tool = True
return self._stream_response_format_fragment(tool_use, self.response_format_schema), None
# Convert tool to content if we're tracking a response_format tool
if self.is_response_format_tool:
message: Final = AnthropicConfig.convert_tool_response_to_message(tool_calls=[tool_use])
@ -1083,6 +1129,45 @@ class ModelResponseIterator:
return text, tool_use
def _stream_response_format_fragment(
self, tool_use: ChatCompletionToolCallChunk, schema: ResponseFormatSchema
) -> str:
if tool_use.get("id") is not None:
self._response_format_buffer = ""
self._response_format_verdict = "undecided"
return ""
fragment: Final = tool_use.get("function", {}).get("arguments") or ""
if self._response_format_verdict == "pass_through":
return fragment
self._response_format_buffer = self._response_format_buffer + fragment
if self._response_format_verdict == "hold":
return ""
self._response_format_verdict = _envelope_verdict(
self._response_format_buffer, schema, RESPONSE_FORMAT_TOOL_NAME
)
return self._response_format_buffer if self._response_format_verdict == "pass_through" else ""
def _flush_response_format_buffer(self) -> str:
if self.response_format_schema is None or self._response_format_verdict == "pass_through":
return ""
held: Final = self._response_format_buffer
self._response_format_buffer = ""
self._response_format_verdict = "undecided"
if not held:
return ""
message: Final = AnthropicConfig.convert_tool_response_to_message(
tool_calls=[
ChatCompletionToolCallChunk(
id=None,
type="function",
function=ChatCompletionToolCallFunctionChunk(name=None, arguments=held),
index=self.tool_index,
)
],
response_format_schema=self.response_format_schema,
)
return (message.content if message is not None else None) or held
def _handle_message_delta(self, chunk: dict) -> tuple[str, Usage | None, dict[str, Any] | None]:
"""
Handle message_delta event for finish_reason, usage, and container.

View file

@ -38,6 +38,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
from litellm.llms.base_llm.base_utils import (
ResponseFormatSchema,
as_json_object,
response_format_schema,
type_to_response_format_param,
unwrap_response_format_envelope,
)
@ -1751,7 +1752,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
else:
_input_schema.update(cast(AnthropicInputSchema, json_schema))
_tool: Final = AnthropicMessagesTool(name=RESPONSE_FORMAT_TOOL_NAME, input_schema=_input_schema)
_tool: Final = AnthropicMessagesTool(
name=RESPONSE_FORMAT_TOOL_NAME,
description="Return the final answer as this structured object, filling every required field.",
input_schema=_input_schema,
)
return _tool
def should_strip_billing_metadata(self) -> bool:
@ -2835,11 +2840,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
prefix_prompt=prefix_prompt,
speed=speed,
tool_name_reverse_map=tool_name_reverse_map,
response_format_schema=(
_response_format_tool_schema(_ResponseFormatRequest.model_validate(optional_params))
if json_mode
else None
),
response_format_schema=JsonModeRequest.model_validate(optional_params).tool_schema() if json_mode else None,
)
return model_response
@ -2861,17 +2862,17 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
except json.JSONDecodeError:
# json decode error does occur, return the original tool response str
return litellm.Message(content=json_mode_content_str)
payload_object: Final = as_json_object(payload)
if payload_object is not None and (values := payload_object.get("values")) is not None:
return litellm.Message(content=json.dumps(values))
if response_format_schema is not None:
return litellm.Message(
content=json.dumps(
unwrap_response_format_envelope(payload, response_format_schema, RESPONSE_FORMAT_TOOL_NAME)
)
)
# a lot of the times the `values` key is not present in the tool response
# relevant issue: https://github.com/BerriAI/litellm/issues/6741
content: Final = (
payload
if response_format_schema is None
else unwrap_response_format_envelope(payload, response_format_schema)
)
return litellm.Message(content=json.dumps(content))
payload_object: Final = as_json_object(payload)
values: Final = None if payload_object is None else payload_object.get("values")
return litellm.Message(content=json.dumps(payload if values is None else values))
@classmethod
def convert_tool_response_to_message(
@ -2889,22 +2890,21 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
class _ResponseFormatRequest(BaseModel):
_REQUEST_TOOLS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
class JsonModeRequest(BaseModel):
model_config = ConfigDict(frozen=True)
tools: object = None
_REQUEST_TOOLS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
def _response_format_tool_schema(request: _ResponseFormatRequest) -> ResponseFormatSchema | None:
try:
tools: Final = _REQUEST_TOOLS.validate_python(request.tools)
def tool_schema(self) -> ResponseFormatSchema | None:
try:
tools: Final = _REQUEST_TOOLS.validate_python(self.tools)
except ValidationError:
return None
json_tool: Final = next((tool for tool in tools if tool.get("name") == RESPONSE_FORMAT_TOOL_NAME), None)
return None if json_tool is None else ResponseFormatSchema.model_validate(json_tool.get("input_schema"))
except ValidationError:
return None
return None if json_tool is None else response_format_schema(json_tool.get("input_schema"))
def _valid_user_id(user_id: str) -> bool:

View file

@ -11,6 +11,7 @@ import httpx
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.chat.handler import AnthropicChatCompletion
from litellm.llms.anthropic.chat.transformation import JsonModeRequest
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
HTTPHandler,
@ -162,6 +163,9 @@ class AzureAnthropicChatCompletion(AnthropicChatCompletion):
logging_obj=logging_obj,
timeout=timeout,
json_mode=json_mode,
response_format_schema=(
JsonModeRequest.model_validate(optional_params).tool_schema() if json_mode else None
),
)
from litellm.llms.anthropic.common_utils import (
process_anthropic_headers,

View file

@ -6,10 +6,11 @@ import copy
import json
from abc import ABC, abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any, Final
from openai.lib import _parsing, _pydantic # pyright: ignore[reportPrivateUsage] # SDK parser internals
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError
from litellm._logging import verbose_logger
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolCallChunk
@ -157,37 +158,81 @@ def convert_tool_response_to_message(
_convert_tool_response_to_message = convert_tool_response_to_message
class ResponseFormatSchema(BaseModel):
model_config = ConfigDict(frozen=True)
@dataclass(frozen=True, slots=True)
class ResponseFormatSchema:
document: Mapping[str, JsonValue]
property_names: frozenset[str]
required: tuple[str, ...]
properties: Mapping[str, object] | None = None
required: tuple[str, ...] = ()
def accepts(self, value: Mapping[str, JsonValue]) -> bool:
from jsonschema.validators import validator_for
from referencing import Registry
from referencing.exceptions import Unresolvable
validator: Final = validator_for(self.document)(dict(self.document), registry=Registry())
try:
return validator.is_valid(dict(value))
except Unresolvable:
return False
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object])
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, JsonValue])
_REQUIRED_NAMES: Final = TypeAdapter(tuple[str, ...])
def as_json_object(value: object) -> Mapping[str, object] | None:
def as_json_object(value: object) -> Mapping[str, JsonValue] | None:
try:
return _JSON_OBJECT.validate_python(value)
except ValidationError:
return None
def _is_schema_object(
value: Mapping[str, object], schema: ResponseFormatSchema, property_names: frozenset[str]
) -> bool:
return not property_names.isdisjoint(value) and all(key in value for key in schema.required)
def response_format_schema(value: object) -> ResponseFormatSchema | None:
document: Final = as_json_object(value)
if document is None:
return None
try:
required: Final = _REQUIRED_NAMES.validate_python(document.get("required", ()))
except ValidationError:
return None
return ResponseFormatSchema(
document=document,
property_names=frozenset(as_json_object(document.get("properties")) or ()),
required=required,
)
def unwrap_response_format_envelope(payload: object, schema: ResponseFormatSchema) -> object:
property_names: Final = frozenset(schema.properties or ())
def _is_schema_object(value: Mapping[str, JsonValue], schema: ResponseFormatSchema) -> bool:
return not schema.property_names.isdisjoint(value) and all(key in value for key in schema.required)
RESPONSE_FORMAT_ENVELOPE_NAME_KEYS: Final = frozenset({"name", "tool_name"})
def _tool_call_envelope_payload(
payload_object: Mapping[str, JsonValue], tool_name: str
) -> Mapping[str, JsonValue] | None:
name_keys: Final = frozenset(
key for key in RESPONSE_FORMAT_ENVELOPE_NAME_KEYS if payload_object.get(key) == tool_name
)
other_values: Final = tuple(as_json_object(value) for key, value in payload_object.items() if key not in name_keys)
if not name_keys or len(other_values) != 1:
return None
return other_values[0]
def unwrap_response_format_envelope(payload: object, schema: ResponseFormatSchema, tool_name: str) -> object:
payload_object: Final = as_json_object(payload)
if payload_object is None or _is_schema_object(payload_object, schema, property_names):
if payload_object is None:
return payload
tool_call_payload: Final = _tool_call_envelope_payload(payload_object, tool_name)
if tool_call_payload is not None:
return tool_call_payload
if schema.accepts(payload_object) or _is_schema_object(payload_object, schema):
return payload
nested_objects: Final = tuple(as_json_object(value) for value in payload_object.values())
candidates: Final = tuple(
nested for nested in nested_objects if nested is not None and _is_schema_object(nested, schema, property_names)
nested for nested in nested_objects if nested is not None and _is_schema_object(nested, schema)
)
return candidates[0] if len(candidates) == 1 else payload

View file

@ -1,9 +1,10 @@
import json
import types
from collections.abc import AsyncIterator, Iterator, Mapping
from typing import TYPE_CHECKING, Final, cast
import httpx
from pydantic import TypeAdapter
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm import verbose_logger
@ -16,6 +17,8 @@ from litellm.litellm_core_utils.optional_imports import ensure_optional_import
from litellm.llms.anthropic.chat.handler import (
ModelResponseIterator as AnthropicModelResponseIterator,
)
from litellm.llms.anthropic.chat.transformation import JsonModeRequest
from litellm.llms.base_llm.base_utils import ResponseFormatSchema
from litellm.llms.custom_httpx.http_handler import (
AsyncHTTPHandler,
HTTPHandler,
@ -174,15 +177,34 @@ class AmazonCohereChatConfig:
return optional_params
_REQUEST_BODY: Final = TypeAdapter(Mapping[str, object])
def response_format_schema_from_request(data: str, json_mode: bool | None) -> ResponseFormatSchema | None:
if not json_mode:
return None
try:
body: Final = _REQUEST_BODY.validate_python(json.loads(data))
except (json.JSONDecodeError, ValidationError):
return None
return JsonModeRequest.model_validate(body).tool_schema()
def _stream_decoder(
bedrock_invoke_provider: litellm.BEDROCK_INVOKE_PROVIDERS_LITERAL | None,
*,
model: str,
json_mode: bool | None,
sync_stream: bool,
response_format_schema: ResponseFormatSchema | None = None,
) -> "AWSEventStreamDecoder":
if bedrock_invoke_provider == "anthropic":
return AmazonAnthropicClaudeStreamDecoder(model=model, sync_stream=sync_stream, json_mode=json_mode)
return AmazonAnthropicClaudeStreamDecoder(
model=model,
sync_stream=sync_stream,
json_mode=json_mode,
response_format_schema=response_format_schema,
)
if bedrock_invoke_provider == "deepseek_r1":
return AmazonDeepSeekR1StreamDecoder(model=model, sync_stream=sync_stream)
if bedrock_invoke_provider == "moonshot":
@ -250,7 +272,11 @@ async def make_call(
)
else:
decoder: Final = _stream_decoder(
bedrock_invoke_provider, model=model, json_mode=json_mode, sync_stream=False
bedrock_invoke_provider,
model=model,
json_mode=json_mode,
sync_stream=False,
response_format_schema=response_format_schema_from_request(data, json_mode),
)
completion_stream = decoder.aiter_bytes(
response.aiter_bytes(chunk_size=stream_chunk_size), response_headers=response.headers
@ -341,7 +367,11 @@ def make_sync_call(
)
else:
decoder: Final = _stream_decoder(
bedrock_invoke_provider, model=model, json_mode=json_mode, sync_stream=True
bedrock_invoke_provider,
model=model,
json_mode=json_mode,
sync_stream=True,
response_format_schema=response_format_schema_from_request(data, json_mode),
)
completion_stream = decoder.iter_bytes(
response.iter_bytes(chunk_size=stream_chunk_size), response_headers=response.headers
@ -878,6 +908,7 @@ class AmazonAnthropicClaudeStreamDecoder(AWSEventStreamDecoder):
model: str,
sync_stream: bool,
json_mode: bool | None = None,
response_format_schema: ResponseFormatSchema | None = None,
) -> None:
"""
Child class of AWSEventStreamDecoder that handles the streaming response from the Anthropic family of models
@ -889,6 +920,7 @@ class AmazonAnthropicClaudeStreamDecoder(AWSEventStreamDecoder):
streaming_response=None,
sync_stream=sync_stream,
json_mode=json_mode,
response_format_schema=response_format_schema,
)
def _chunk_parser(self, chunk_data: dict) -> ModelResponseStream:

View file

@ -47,7 +47,7 @@ def _as_output_config(value: object) -> Mapping[str, object] | None:
def _fold_output_format_into_output_config(
output_config: Mapping[str, object] | None, output_format: object
) -> Mapping[str, object] | None:
if output_format is None or (output_config is not None and "format" in output_config):
if output_format is None or (output_config is not None and output_config.get("format") is not None):
return output_config
return {**(output_config or {}), "format": output_format}

File diff suppressed because it is too large Load diff

View file

@ -12,6 +12,7 @@ from litellm._uuid import uuid
from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
from litellm.litellm_core_utils.prompt_templates.factory import anthropic_messages_pt
from litellm.llms.anthropic.chat.handler import ModelResponseIterator, make_call
from litellm.llms.base_llm.base_utils import response_format_schema
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.llms.openai import (
ChatCompletionToolCallChunk,
@ -2367,3 +2368,112 @@ def test_served_model_reaches_assembled_stream_through_custom_stream_wrapper():
assert chunk._hidden_params["provider_response_model"] == served_model
assembled: Final = litellm.stream_chunk_builder(chunks=list(chunks), messages=[{"role": "user", "content": "hi"}])
assert assembled._hidden_params["provider_response_model"] == served_model
_STREAMED_RISK_SCHEMA: Final = {
"type": "object",
"properties": {"tier": {"type": "integer"}, "summary": {"type": "string"}},
"required": ["tier", "summary"],
}
_STREAMED_RISK_OBJECT: Final = {"tier": 3, "summary": "Two-hour read-only window on the loyalty APIs"}
_STREAMED_LOOSE_SCHEMA: Final = {
"type": "object",
"properties": {"name": {"type": "string"}, "email": {"type": "string"}},
}
_STREAMED_CONTACT: Final = {"name": "Ada", "email": "ada@example.com"}
def _json_tool_events(fragments: tuple[str, ...]) -> tuple[dict[str, object], ...]:
return (
{
"type": "message_start",
"message": {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-emulated-structured-output",
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 10, "output_tokens": 1},
},
},
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "tool_use", "id": "toolu_1", "name": RESPONSE_FORMAT_TOOL_NAME, "input": {}},
},
*(
{"type": "content_block_delta", "index": 0, "delta": {"type": "input_json_delta", "partial_json": fragment}}
for fragment in fragments
),
{"type": "content_block_stop", "index": 0},
{
"type": "message_delta",
"delta": {"stop_reason": "tool_use", "stop_sequence": None},
"usage": {"output_tokens": 20},
},
)
def _streamed_json_mode_contents(
fragments: tuple[str, ...], schema: dict[str, object] = _STREAMED_RISK_SCHEMA
) -> tuple[str, ...]:
iterator = ModelResponseIterator(
streaming_response=MagicMock(),
sync_stream=True,
json_mode=True,
response_format_schema=response_format_schema(schema),
)
chunks = tuple(iterator.chunk_parser(event) for event in _json_tool_events(fragments))
assert all(chunk.choices[0].delta.tool_calls is None for chunk in chunks)
assert chunks[-1].choices[0].finish_reason == "stop"
return tuple(chunk.choices[0].delta.content or "" for chunk in chunks)
def test_streamed_json_mode_envelope_is_unwrapped_when_the_block_ends():
envelope = json.dumps({"name": RESPONSE_FORMAT_TOOL_NAME, "parameters": _STREAMED_RISK_OBJECT})
fragments = (envelope[:9], envelope[9:40], envelope[40:])
contents = _streamed_json_mode_contents(fragments)
assert json.loads("".join(contents)) == _STREAMED_RISK_OBJECT
assert contents[:-2] == ("",) * (len(contents) - 2)
def test_streamed_json_mode_schema_object_streams_as_it_arrives():
answer = json.dumps(_STREAMED_RISK_OBJECT)
fragments = (answer[:4], answer[4:12], answer[12:])
contents = _streamed_json_mode_contents(fragments)
assert "".join(contents) == answer
assert contents[3] == answer[:12]
assert contents[4] == answer[12:]
assert contents[5] == ""
def test_streamed_json_mode_truncated_output_is_passed_through_untouched():
contents = _streamed_json_mode_contents(('{"na', 'me": "json_tool_call", "parameters": {"tier": 3'))
assert "".join(contents) == '{"name": "json_tool_call", "parameters": {"tier": 3'
def test_streamed_json_mode_tool_name_envelope_is_unwrapped_when_name_is_a_schema_property():
envelope = json.dumps({"name": RESPONSE_FORMAT_TOOL_NAME, "parameters": _STREAMED_CONTACT})
contents = _streamed_json_mode_contents((envelope[:6], envelope[6:24], envelope[24:]), _STREAMED_LOOSE_SCHEMA)
assert json.loads("".join(contents)) == _STREAMED_CONTACT
assert contents[:-2] == ("",) * (len(contents) - 2)
def test_streamed_json_mode_name_property_answer_streams_once_its_value_is_known():
answer = json.dumps(_STREAMED_CONTACT)
contents = _streamed_json_mode_contents((answer[:6], answer[6:16], answer[16:]), _STREAMED_LOOSE_SCHEMA)
assert "".join(contents) == answer
assert contents[2] == ""
assert contents[3] == answer[:16]
assert contents[4] == answer[16:]

View file

@ -23,7 +23,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import anthropic_messag
from litellm.llms.anthropic.chat import ModelResponseIterator
from litellm.llms.anthropic.common_utils import process_anthropic_headers
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.base_llm.base_utils import ResponseFormatSchema
from litellm.llms.base_llm.base_utils import ResponseFormatSchema, response_format_schema
from litellm.llms.anthropic.pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
@ -7278,12 +7278,30 @@ _PERSON_SCHEMA: Final = {
"properties": {"name": {"type": "string"}, "age": {"type": "integer"}},
"required": ["name", "age"],
}
_OPTIONAL_FIELDS_SCHEMA: Final = {"type": "object", "properties": {"note": {"type": "string"}, "score": {"type": "integer"}}}
_OPTIONAL_FIELDS_SCHEMA: Final = {
"type": "object",
"properties": {"note": {"type": "string"}, "score": {"type": "integer"}},
}
_TREE_SCHEMA: Final = {
"type": "object",
"properties": {"value": {"type": "integer"}, "child": {"type": "object"}},
"required": ["value"],
}
_OPEN_SCORE_SCHEMA: Final = {
"type": "object",
"properties": {"score": {"type": "integer"}},
"additionalProperties": True,
}
_VALUES_SCHEMA: Final = {"type": "object", "properties": {"values": {"type": "array"}}, "required": ["values"]}
_REMOTE_REF_SCHEMA: Final = {**_PERSON_SCHEMA, "$ref": "https://schemas.example.com/person.json"}
_LOOSE_SCHEMA: Final = {"type": "object", "properties": {"name": {"type": "string"}, "email": {"type": "string"}}}
_CONTACT: Final = {"name": "Ada", "email": "ada@example.com"}
def _schema(document: dict[str, object]) -> ResponseFormatSchema:
schema = response_format_schema(document)
assert schema is not None
return schema
def _json_tool_call(arguments: object) -> list[ChatCompletionToolCallChunk]:
@ -7291,7 +7309,9 @@ def _json_tool_call(arguments: object) -> list[ChatCompletionToolCallChunk]:
ChatCompletionToolCallChunk(
id="toolu_1",
type="function",
function=ChatCompletionToolCallFunctionChunk(name=RESPONSE_FORMAT_TOOL_NAME, arguments=json.dumps(arguments)),
function=ChatCompletionToolCallFunctionChunk(
name=RESPONSE_FORMAT_TOOL_NAME, arguments=json.dumps(arguments)
),
index=0,
)
]
@ -7314,7 +7334,37 @@ def _json_tool_call(arguments: object) -> list[ChatCompletionToolCallChunk]:
{"name": "Ada", "age": 36},
id="envelope_key_shared_with_a_schema_property",
),
pytest.param(_OPTIONAL_FIELDS_SCHEMA, {"input": {"score": 7}}, {"score": 7}, id="optional_fields_wrapper"),
pytest.param(
_OPTIONAL_FIELDS_SCHEMA,
{"input": {"score": 7}},
{"input": {"score": 7}},
id="wrapper_the_schema_itself_accepts",
),
pytest.param(
_OPEN_SCORE_SCHEMA,
{"meta": {"score": 7}, "label": "keep"},
{"meta": {"score": 7}, "label": "keep"},
id="valid_answer_with_extra_keys_and_a_schema_shaped_nested_object",
),
pytest.param(_VALUES_SCHEMA, {"values": [1, 2]}, {"values": [1, 2]}, id="values_is_a_schema_property"),
pytest.param(
_REMOTE_REF_SCHEMA,
{"input": {"name": "Ada", "age": 36}},
{"name": "Ada", "age": 36},
id="remote_ref_schema_falls_back_to_key_matching",
),
pytest.param(
_LOOSE_SCHEMA,
{"name": RESPONSE_FORMAT_TOOL_NAME, "parameters": _CONTACT},
_CONTACT,
id="tool_name_envelope_when_the_schema_lists_no_required_keys",
),
pytest.param(
_LOOSE_SCHEMA,
{"name": RESPONSE_FORMAT_TOOL_NAME, "email": "ada@example.com"},
{"name": RESPONSE_FORMAT_TOOL_NAME, "email": "ada@example.com"},
id="tool_name_value_without_a_nested_object_stays",
),
pytest.param(_RISK_SCHEMA, _RISK_OBJECT, _RISK_OBJECT, id="plain_schema_object"),
pytest.param(
_TREE_SCHEMA,
@ -7322,7 +7372,9 @@ def _json_tool_call(arguments: object) -> list[ChatCompletionToolCallChunk]:
{"value": 1, "child": {"value": 2}},
id="schema_object_whose_property_holds_a_schema_shaped_object",
),
pytest.param(_RISK_SCHEMA, {"input": {"tier": 3}}, {"input": {"tier": 3}}, id="inner_object_missing_required_key"),
pytest.param(
_RISK_SCHEMA, {"input": {"tier": 3}}, {"input": {"tier": 3}}, id="inner_object_missing_required_key"
),
pytest.param(
_RISK_SCHEMA,
{"first": _RISK_OBJECT, "second": _RISK_OBJECT},
@ -7348,7 +7400,7 @@ def test_convert_tool_response_to_message_unwraps_only_an_unambiguous_schema_env
):
message = AnthropicConfig.convert_tool_response_to_message(
tool_calls=_json_tool_call(tool_input),
response_format_schema=ResponseFormatSchema.model_validate(schema),
response_format_schema=_schema(schema),
)
assert message is not None
@ -7366,7 +7418,7 @@ def test_json_mode_with_user_tool_call_unwraps_the_envelope_into_content():
replacement, remaining_tool_calls, extra_content = AnthropicConfig()._resolve_json_mode_non_streaming(
json_mode=True,
tool_calls=[*_json_tool_call({"input": _RISK_OBJECT}), user_tool_call],
response_format_schema=ResponseFormatSchema.model_validate(_RISK_SCHEMA),
response_format_schema=_schema(_RISK_SCHEMA),
)
assert replacement is None
@ -7402,6 +7454,18 @@ def test_vertex_claude_tool_emulated_response_format_returns_the_object_the_mode
drop_params=False,
)
json_mode = optional_params.pop("json_mode")
request = config.transform_request(
model="claude-emulated-structured-output",
messages=[{"role": "user", "content": "Assess this change request"}],
optional_params={**optional_params, "is_vertex_request": True},
litellm_params={},
headers={},
)
json_tool = next(tool for tool in request["tools"] if tool["name"] == RESPONSE_FORMAT_TOOL_NAME)
assert json_tool["input_schema"]["required"] == ["tier", "summary"]
assert json_tool["description"]
assert "output_format" not in request
assert "output_config" not in request
completion_response = {
"id": "msg_1",
"type": "message",
@ -7425,7 +7489,7 @@ def test_vertex_claude_tool_emulated_response_format_returns_the_object_the_mode
raw_response=httpx.Response(200, json=completion_response),
model_response=ModelResponse(),
logging_obj=MagicMock(),
request_data={},
request_data=request,
messages=[{"role": "user", "content": "Assess this change request"}],
optional_params=optional_params,
litellm_params={},
@ -7749,9 +7813,7 @@ async def test_anthropic_api_max_completion_tokens(model: str):
print("request_body: ", request_body)
assert request_body == {
"messages": [
{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}
],
"messages": [{"role": "user", "content": [{"type": "text", "text": "Hello!"}]}],
"max_tokens": 10,
"model": model.split("/")[-1],
}
@ -7969,9 +8031,7 @@ async def test_anthropic_api_prompt_caching_with_content_str():
"cache_control": {"type": "ephemeral"},
},
]
translated_system_message = litellm.AnthropicConfig().translate_system_message(
messages=system_message
)
translated_system_message = litellm.AnthropicConfig().translate_system_message(messages=system_message)
assert translated_system_message == [
# System Message
@ -8041,9 +8101,9 @@ async def test_anthropic_api_prompt_caching_with_content_str():
assert len(translated_messages) == len(expected_messages)
for idx, i in enumerate(translated_messages):
assert (
i == expected_messages[idx]
), "Error on idx={}. Got={}, Expected={}".format(idx, i, expected_messages[idx])
assert i == expected_messages[idx], "Error on idx={}. Got={}, Expected={}".format(
idx, i, expected_messages[idx]
)
@pytest.fixture

View file

@ -20,8 +20,10 @@ from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.llms.bedrock.chat.invoke_handler import (
AmazonOpenAICompatibleStreamDecoder,
AWSEventStreamDecoder,
_stream_decoder,
make_call,
make_sync_call,
response_format_schema_from_request,
)
from litellm.llms.bedrock.common_utils import BedrockError, get_bedrock_stream_event_statuses
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
@ -1442,3 +1444,88 @@ def test_invoke_decoder_reports_only_missing_aws_dependency(missing):
assert error.value.__cause__ is failure
else:
assert error.value is failure
_INVOKE_RISK_SCHEMA: Final = {
"type": "object",
"properties": {"tier": {"type": "integer"}, "summary": {"type": "string"}},
"required": ["tier", "summary"],
}
_INVOKE_RISK_OBJECT: Final = {"tier": 2, "summary": "Weekend database migration with a tested rollback"}
def _invoke_json_mode_request(tools: Sequence[Mapping[str, object]]) -> str:
return json.dumps(
{
"anthropic_version": "bedrock-2023-05-31",
"messages": [{"role": "user", "content": "Assess this change request"}],
"tools": list(tools),
"tool_choice": {"type": "tool", "name": "json_tool_call"},
}
)
def _invoke_json_tool_events(fragments: Sequence[str]) -> tuple[dict[str, object], ...]:
deltas = tuple(
{"type": "content_block_delta", "index": 0, "delta": {"type": "input_json_delta", "partial_json": fragment}}
for fragment in fragments
)
return (
{"type": "message_start", "message": {"id": "msg_1", "usage": {"input_tokens": 1, "output_tokens": 0}}},
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "tool_use", "id": "t1", "name": "json_tool_call", "input": {}},
},
*deltas,
{"type": "content_block_stop", "index": 0},
{"type": "message_delta", "delta": {"stop_reason": "tool_use"}, "usage": {"output_tokens": 9}},
)
def _invoke_streamed_json_mode_contents(data: str, fragments: Sequence[str], json_mode: bool = True) -> tuple[str, ...]:
decoder = _stream_decoder(
"anthropic",
model="anthropic.claude-sonnet-4",
json_mode=True,
sync_stream=True,
response_format_schema=response_format_schema_from_request(data, json_mode),
)
chunks = tuple(decoder._chunk_parser(event) for event in _invoke_json_tool_events(fragments))
assert all(not chunk.choices[0].delta.tool_calls for chunk in chunks)
return tuple(chunk.choices[0].delta.content or "" for chunk in chunks)
def test_invoke_streamed_json_mode_envelope_is_unwrapped_with_the_schema_from_the_request_body():
envelope: Final = json.dumps({"tool_name": "json_tool_call", "parameters": _INVOKE_RISK_OBJECT})
data: Final = _invoke_json_mode_request([{"name": "json_tool_call", "input_schema": _INVOKE_RISK_SCHEMA}])
contents = _invoke_streamed_json_mode_contents(data, (envelope[:17], envelope[17:48], envelope[48:]))
assert json.loads("".join(contents)) == _INVOKE_RISK_OBJECT
@pytest.mark.parametrize(
("data", "json_mode"),
[
pytest.param("not json", True, id="unparseable_body"),
pytest.param(json.dumps([1, 2]), True, id="body_is_not_an_object"),
pytest.param(
_invoke_json_mode_request([{"name": "other_tool", "input_schema": _INVOKE_RISK_SCHEMA}]),
True,
id="no_json_tool",
),
pytest.param(
_invoke_json_mode_request([{"name": "json_tool_call", "input_schema": _INVOKE_RISK_SCHEMA}]),
False,
id="json_mode_off",
),
],
)
def test_invoke_stream_keeps_the_wrapper_when_the_request_body_carries_no_json_tool_schema(data: str, json_mode: bool):
envelope: Final = json.dumps({"tool_name": "json_tool_call", "parameters": _INVOKE_RISK_OBJECT})
assert response_format_schema_from_request(data, json_mode) is None
contents = _invoke_streamed_json_mode_contents(data, (envelope[:17], envelope[17:]), json_mode)
assert json.loads("".join(contents)) == {"tool_name": "json_tool_call", "parameters": _INVOKE_RISK_OBJECT}

View file

@ -787,6 +787,12 @@ _CALLER_FORMAT: Final = {
{"format": _CALLER_FORMAT},
id="caller_output_config_format_wins",
),
pytest.param(
"claude-opus-4-6",
{"output_format": _STRUCTURED_FORMAT, "output_config": {"format": None}},
{"format": _STRUCTURED_FORMAT},
id="null_output_config_format_counts_as_absent",
),
pytest.param(
"claude-opus-4-6",
{"output_format": _STRUCTURED_FORMAT, "output_config": "not-a-dict"},