diff --git a/litellm-rust/crates/core-utils/src/core_helpers.rs b/litellm-rust/crates/core-utils/src/core_helpers.rs index 58ee88d301c..b93f0484bb9 100644 --- a/litellm-rust/crates/core-utils/src/core_helpers.rs +++ b/litellm-rust/crates/core-utils/src/core_helpers.rs @@ -13,6 +13,7 @@ const FINISH_REASONS: &[(&str, &str)] = &[ ("max_tokens", "length"), ("refusal", "content_filter"), ("compaction", "length"), + ("pause_turn", "length"), ("model_context_window_exceeded", "length"), ("guardrail_intervened", "content_filter"), ("content_filtered", "content_filter"), @@ -75,6 +76,7 @@ mod tests { #[case::end_turn("end_turn", "stop")] #[case::stop_sequence("stop_sequence", "stop")] #[case::max_tokens("max_tokens", "length")] + #[case::pause_turn("pause_turn", "length")] #[case::model_context_window_exceeded("model_context_window_exceeded", "length")] #[case::refusal("refusal", "content_filter")] #[case::guardrail_intervened("guardrail_intervened", "content_filter")] diff --git a/litellm/litellm_core_utils/core_helpers.py b/litellm/litellm_core_utils/core_helpers.py index 20e10f63f83..00151107213 100644 --- a/litellm/litellm_core_utils/core_helpers.py +++ b/litellm/litellm_core_utils/core_helpers.py @@ -311,6 +311,7 @@ _FINISH_REASON_MAP: Final[dict[str, OpenAIChatCompletionFinishReason]] = { "tool_use": "tool_calls", "refusal": "content_filter", "compaction": "length", + "pause_turn": "length", "model_context_window_exceeded": "length", # Cohere "COMPLETE": "stop", diff --git a/litellm/litellm_core_utils/prompt_templates/factory.py b/litellm/litellm_core_utils/prompt_templates/factory.py index de896cde4dc..0bbb25163cd 100644 --- a/litellm/litellm_core_utils/prompt_templates/factory.py +++ b/litellm/litellm_core_utils/prompt_templates/factory.py @@ -26,6 +26,7 @@ from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_or from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import ( anthropic_system_messages, find_anthropic_server_tool_result, + is_anthropic_server_tool_use, is_anthropic_tool_search_use, message_field, parts_of, @@ -2059,6 +2060,54 @@ def _placed_and_paired( _AnthropicToolInvokeItem: TypeAlias = AnthropicMessagesToolUseParam | dict[str, Any] +def _is_server_tool_result(block: Mapping[str, object]) -> bool: + block_type: Final = block.get("type") + return isinstance(block_type, str) and block_type.endswith("_tool_result") + + +def _split_unrun_server_tool_uses( + groups: Sequence[Sequence[_AnthropicToolInvokeItem]], +) -> tuple[tuple[Sequence[_AnthropicToolInvokeItem], ...], tuple[_AnthropicToolInvokeItem, ...]]: + completed: Final = tuple(group for group in groups if any(_is_server_tool_result(block) for block in group)) + unrun: Final = tuple(group[0] for group in groups if not any(_is_server_tool_result(block) for block in group)) + return completed, unrun + + +def _block_id(block: Mapping[str, object]) -> object: + return block.get("id") + + +def _unrun_server_tool_use_ids(blocks: Sequence[Mapping[str, object]]) -> frozenset[object]: + result_ids: Final = frozenset(block.get("tool_use_id") for block in blocks if _is_server_tool_result(block)) + return frozenset( + _block_id(block) + for block in blocks + if block.get("type") == "server_tool_use" and _block_id(block) not in result_ids + ) + + +def _last_block_id_among(blocks: Sequence[Mapping[str, object]], ids: frozenset[object]) -> object: + return next(block_id for block in reversed(blocks) if (block_id := _block_id(block)) in ids) + + +def _paused_turn_replay( + tool_invokes: Sequence[_AnthropicToolInvokeItem], + thinking_blocks: Sequence[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock], + *, + is_final_message: bool, +) -> tuple[ + tuple[_AnthropicToolInvokeItem, ...], + tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...], +]: + unrun_ids: Final = _unrun_server_tool_use_ids(tool_invokes) + if not unrun_ids: + return tuple(tool_invokes), tuple(thinking_blocks) + dropped_ids: Final = ( + unrun_ids - frozenset((_last_block_id_among(tool_invokes, unrun_ids),)) if is_final_message else unrun_ids + ) + return tuple(block for block in tool_invokes if _block_id(block) not in dropped_ids), () + + def convert_to_anthropic_tool_invoke( tool_calls: list[ChatCompletionAssistantToolCall], web_search_results: Sequence[object] | None = None, @@ -2141,6 +2190,15 @@ def convert_to_anthropic_tool_invoke( anthropic_tool_invoke.append(server_tool_result) elif is_anthropic_tool_search_use(tool_id, tool_name) and tool_id not in answered_tool_ids: continue + elif is_anthropic_server_tool_use(tool_id) and tool_id not in answered_tool_ids: + anthropic_tool_invoke.append( + { + "type": "server_tool_use", + "id": tool_id, + "name": tool_name, + "input": tool_input, + } + ) else: sanitized_tool_id = sanitize_anthropic_tool_use_id(tool_id) _anthropic_tool_use_param = AnthropicMessagesToolUseParam( @@ -3109,6 +3167,30 @@ def anthropic_messages_pt( _has_server_tool_calls = True break + # Get web_search_results and tool_results from provider_specific_fields + # for server_tool_use reconstruction. + # Fixes: https://github.com/BerriAI/litellm/issues/17737 + _provider_specific_fields_raw = assistant_content_block.get("provider_specific_fields") + _provider_specific_fields: dict[str, Any] = ( + cast(dict[str, Any], _provider_specific_fields_raw) + if isinstance(_provider_specific_fields_raw, dict) + else {} + ) + tool_invoke_results = ( + convert_to_anthropic_tool_invoke( + assistant_tool_calls, + web_search_results=_provider_specific_fields.get("web_search_results"), + tool_results=_provider_specific_fields.get("tool_results"), + answered_tool_ids=client_answered_tool_ids, + ) + if assistant_tool_calls is not None + else () + ) + turn_paused = bool(_unrun_server_tool_use_ids(tool_invoke_results)) + replayed_tool_invokes, replayed_thinking_blocks = _paused_turn_replay( + tool_invoke_results, thinking_blocks or (), is_final_message=msg_i == len(messages) - 1 + ) + if ( thinking_blocks is not None and _has_server_tool_calls @@ -3121,26 +3203,12 @@ def anthropic_messages_pt( # We must preserve this interleaved order because Anthropic # verifies thinking block signatures based on position. - # Build the tool call groups (server_tool_use + its result) - _provider_specific_fields_raw_tc = assistant_content_block.get("provider_specific_fields") - _provider_specific_fields_tc: dict[str, Any] = {} - if isinstance(_provider_specific_fields_raw_tc, dict): - _provider_specific_fields_tc = cast(dict[str, Any], _provider_specific_fields_raw_tc) - _web_search_results_tc = _provider_specific_fields_tc.get("web_search_results") - _tool_results_tc = _provider_specific_fields_tc.get("tool_results") - tool_invoke_results = convert_to_anthropic_tool_invoke( - assistant_tool_calls, - web_search_results=_web_search_results_tc, - tool_results=_tool_results_tc, - answered_tool_ids=client_answered_tool_ids, - ) - # Group tool invoke results into (server_tool_use, result) pairs # and separate regular tool_use blocks server_tool_groups: list[list[_AnthropicToolInvokeItem]] = [] regular_tool_uses: list[_AnthropicToolInvokeItem] = [] _current_group: list[_AnthropicToolInvokeItem] = [] - for item in tool_invoke_results: + for item in replayed_tool_invokes: item_type = item.get("type", "") if isinstance(item, dict) else getattr(item, "type", "") if item_type == "server_tool_use": if _current_group: @@ -3154,6 +3222,9 @@ def anthropic_messages_pt( _current_group.append(item) if _current_group: server_tool_groups.append(_current_group) + completed_server_tool_groups, pending_server_tool_uses = _split_unrun_server_tool_uses( + server_tool_groups + ) # Build the text block if content is a non-empty string text_element = None @@ -3177,15 +3248,15 @@ def anthropic_messages_pt( # Any remaining groups (after all thinking blocks) go after. tb_idx = 0 grp_idx = 0 - num_tb = len(thinking_blocks) if thinking_blocks else 0 - num_grp = len(server_tool_groups) + num_tb = len(replayed_thinking_blocks) + num_grp = len(completed_server_tool_groups) while tb_idx < num_tb or grp_idx < num_grp: if tb_idx < num_tb and grp_idx < num_grp: # Emit thinking block then its tool group - assistant_content.append(thinking_blocks[tb_idx]) + assistant_content.append(replayed_thinking_blocks[tb_idx]) tb_idx += 1 - for block in server_tool_groups[grp_idx]: + for block in completed_server_tool_groups[grp_idx]: item_id = block.get("id") if isinstance(block, dict) else getattr(block, "id", None) if item_id and item_id in unique_tool_ids: continue @@ -3195,11 +3266,11 @@ def anthropic_messages_pt( grp_idx += 1 elif tb_idx < num_tb: # More thinking blocks than tool groups - emit before text - assistant_content.append(thinking_blocks[tb_idx]) + assistant_content.append(replayed_thinking_blocks[tb_idx]) tb_idx += 1 else: # More tool groups than thinking blocks - emit remaining - for block in server_tool_groups[grp_idx]: + for block in completed_server_tool_groups[grp_idx]: item_id = block.get("id") if isinstance(block, dict) else getattr(block, "id", None) if item_id and item_id in unique_tool_ids: continue @@ -3212,8 +3283,8 @@ def anthropic_messages_pt( if text_element is not None: assistant_content.append(text_element) - # Add regular (non-server) tool calls at the end - for item in regular_tool_uses: + # Add regular tool calls, then a paused turn's unrun server tool call, at the end + for item in (*regular_tool_uses, *pending_server_tool_uses): item_id = item.get("id") if isinstance(item, dict) else getattr(item, "id", None) if item_id and item_id in unique_tool_ids: continue @@ -3233,18 +3304,18 @@ def anthropic_messages_pt( ) _content_list = assistant_content_block.get("content") if _content_is_list else None - if ( - thinking_blocks is not None and not _content_is_list - ): # IMPORTANT: ADD THIS FIRST, ELSE ANTHROPIC WILL RAISE AN ERROR - assistant_content.extend(thinking_blocks) + if not _content_is_list: # IMPORTANT: ADD THIS FIRST, ELSE ANTHROPIC WILL RAISE AN ERROR + assistant_content.extend(replayed_thinking_blocks) if _content_is_list and _content_list is not None: - for m in _with_missing_thinking_blocks(_content_list, thinking_blocks or ()): + for m in _with_missing_thinking_blocks(_content_list, replayed_thinking_blocks): if not isinstance(m, dict): continue # handle thinking blocks if (thinking_block := _as_thinking_block(m)) is not None: assistant_content.extend( - _replayable_thinking_blocks([thinking_block], drop_params=drop_params) + () + if turn_paused + else _replayable_thinking_blocks([thinking_block], drop_params=drop_params) ) continue text_block = cast(str, m.get("text", "")) @@ -3284,25 +3355,9 @@ def anthropic_messages_pt( assistant_content.append(_anthropic_text_content_element) if assistant_tool_calls is not None: # support assistant tool invoke conversion - # Get web_search_results and tool_results from provider_specific_fields - # for server_tool_use reconstruction. - # Fixes: https://github.com/BerriAI/litellm/issues/17737 - _provider_specific_fields_raw = assistant_content_block.get("provider_specific_fields") - _provider_specific_fields: dict[str, Any] = {} - if isinstance(_provider_specific_fields_raw, dict): - _provider_specific_fields = cast(dict[str, Any], _provider_specific_fields_raw) - _web_search_results = _provider_specific_fields.get("web_search_results") - _tool_results = _provider_specific_fields.get("tool_results") - tool_invoke_results = convert_to_anthropic_tool_invoke( - assistant_tool_calls, - web_search_results=_web_search_results, - tool_results=_tool_results, - answered_tool_ids=client_answered_tool_ids, - ) - # Prevent "tool_use ids must be unique" errors by filtering duplicates # This can happen when merging history that already contains the tool calls - for item in tool_invoke_results: + for item in replayed_tool_invokes: # tool_use items are typically dicts, but handle objects just in case item_id = item.get("id") if isinstance(item, dict) else getattr(item, "id", None) diff --git a/litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py b/litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py index eb3400aee8b..3f956b0614c 100644 --- a/litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py +++ b/litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py @@ -286,6 +286,10 @@ def _separate_thinking_blocks_render(message: object) -> bool: return any(_thinking_block_renders(block) for block in parts_of(message_field(message, "thinking_blocks"))) +def is_anthropic_server_tool_use(tool_id: str) -> bool: + return tool_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX) + + def is_anthropic_tool_search_use(tool_id: str, tool_name: str) -> bool: return tool_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX) and tool_name in ANTHROPIC_TOOL_SEARCH_TOOL_NAMES diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index 4857973d51d..7191b2ce8e8 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -1509,11 +1509,20 @@ def _delta_reasoning_items(delta: object) -> Sequence[object]: return cast(Sequence[object], items) if isinstance(items, list) else () # cast-ok: items forwarded verbatim -def _chunk_reasoning_items(chunk: object) -> Sequence[object]: +def _chunk_first_delta(chunk: object) -> object: choices: Final = _field_of(chunk, "choices") if not isinstance(choices, list) or not choices: - return () - return _delta_reasoning_items(_field_of(cast(Sequence[object], choices)[0], "delta")) # cast-ok: untyped chunk + return None + return _field_of(cast(Sequence[object], choices)[0], "delta") # cast-ok: untyped chunk + + +def _chunk_reasoning_items(chunk: object) -> Sequence[object]: + return _delta_reasoning_items(_chunk_first_delta(chunk)) + + +def _chunk_refusal(chunk: object) -> str: + refusal: Final = _field_of(_chunk_first_delta(chunk), "refusal") + return refusal if isinstance(refusal, str) else "" _NON_TEXT_DELTA_FIELDS: Final = ( @@ -1526,6 +1535,7 @@ _NON_TEXT_DELTA_FIELDS: Final = ( "audio", "images", "provider_specific_fields", + "refusal", ) @@ -1674,6 +1684,9 @@ def _assembled_message_fields(processor: ChunkProcessor, chunks: Sequence) -> It yield "reasoning_content", processor.get_combined_reasoning_content(reasoning_chunks) typed_chunks: Final = cast(Sequence[object], chunks) # cast-ok: chunks are untyped upstream + combined_refusal: Final = "".join(_chunk_refusal(chunk) for chunk in typed_chunks) + if combined_refusal: + yield "refusal", combined_refusal reasoning_items: Final = tuple(chain.from_iterable(_chunk_reasoning_items(chunk) for chunk in typed_chunks)) if reasoning_items: yield "reasoning_items", list(reasoning_items) diff --git a/litellm/llms/anthropic/chat/handler.py b/litellm/llms/anthropic/chat/handler.py index e1436473426..75f88355652 100644 --- a/litellm/llms/anthropic/chat/handler.py +++ b/litellm/llms/anthropic/chat/handler.py @@ -70,7 +70,12 @@ from litellm.types.utils import ( ) from ...base import BaseLLM -from ..common_utils import AnthropicError, process_anthropic_headers, refusal_provider_specific_fields +from ..common_utils import ( + AnthropicError, + process_anthropic_headers, + refusal_params, + stop_reason_provider_specific_fields, +) from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig, JsonModeRequest if TYPE_CHECKING: @@ -926,6 +931,17 @@ class ModelResponseIterator: def _web_search_call_snapshot(self) -> dict[str, object]: return dict(self._web_search_calls) + def _refresh_pending_web_search_call(self, tool_use_id: str) -> bool: + if tool_use_id not in self._web_search_calls: + return False + self._web_search_calls[tool_use_id] = build_web_search_call( + tool_use_id, + self._server_tool_inputs.get(tool_use_id, {}), + {"content": []}, + status="in_progress", + ) + return True + def _complete_web_search_call(self, result: dict[str, object]) -> None: tool_use_id: Final = result.get("tool_use_id") if not isinstance(tool_use_id, str) or tool_use_id not in self._web_search_calls: @@ -1128,6 +1144,8 @@ class ModelResponseIterator: self._server_tool_inputs[self._current_server_tool_id] = json.loads(args) except (json.JSONDecodeError, TypeError): pass + if self._refresh_pending_web_search_call(self._current_server_tool_id): + provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot() self._current_server_tool_id = None if self.is_response_format_tool: text = self._flush_response_format_buffer() @@ -1140,8 +1158,8 @@ class ModelResponseIterator: # These are automatically handled by Anthropic API, we just pass them through pass elif type_chunk == "message_delta": - finish_reason, usage, container, refusal_fields = self._handle_message_delta(chunk) - provider_specific_fields.update(refusal_fields) + finish_reason, usage, container, stop_reason_fields = self._handle_message_delta(chunk) + provider_specific_fields.update(stop_reason_fields) text = self._release_answer_close() + self._drain_split_answer_blocks() if container: provider_specific_fields["container"] = container @@ -1190,6 +1208,7 @@ class ModelResponseIterator: ) text, tool_use = self._handle_json_mode_chunk(text=text, tool_use=tool_use) + refusal_fields: Final = refusal_params(provider_specific_fields) returned_chunk: Final = ModelResponseStream( choices=[ @@ -1201,6 +1220,7 @@ class ModelResponseIterator: provider_specific_fields=(provider_specific_fields if provider_specific_fields else None), thinking_blocks=(thinking_blocks if thinking_blocks else None), reasoning_content=reasoning_content, + **refusal_fields, ), finish_reason=finish_reason, ) @@ -1371,7 +1391,7 @@ class ModelResponseIterator: self._handle_usage(anthropic_usage_chunk=message_delta["usage"]) if "usage" in message_delta else None ) container: Final = message_delta["delta"].get("container") - return finish_reason, usage, container, refusal_provider_specific_fields(message_delta["delta"]) + return finish_reason, usage, container, stop_reason_provider_specific_fields(message_delta["delta"]) def _handle_accumulated_json_chunk(self, data_str: str, is_final: bool = False) -> ModelResponseStream | None: """ diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index 25245c347f4..f03004cec15 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -120,13 +120,14 @@ from litellm.utils import ( ) from ..common_utils import ( - REFUSAL_STOP_FIELDS, + STOP_REASON_FIELDS, AnthropicError, AnthropicModelInfo, eager_input_streaming_flag, process_anthropic_headers, - refusal_provider_specific_fields, + refusal_params, requires_native_compaction_beta, + stop_reason_provider_specific_fields, strip_advisor_blocks_from_messages, tools_marked_for_a_tool_config_point, ) @@ -167,6 +168,8 @@ _DEFER_LOADING_ADAPTER: Final = TypeAdapter(bool) _TOOL_FIELDS_ADAPTER: Final = TypeAdapter(Mapping[str, object]) _TOOLS_ADAPTER: Final = TypeAdapter(list[object]) _LITELLM_PARAMS_ADAPTER: Final = TypeAdapter(Mapping[str, object]) +_RESPONSE_BLOCKS: Final = TypeAdapter(tuple[Mapping[str, object], ...]) +_PENDING_WEB_SEARCH_RESULT: Final = MappingProxyType({"content": ()}) _REQUEST_PARAMS_ADAPTER: Final = TypeAdapter(Mapping[str, object]) _NATIVE_OUTPUT_STRING_FORMATS: Final = frozenset( ("date-time", "time", "date", "duration", "email", "hostname", "uri", "ipv4", "ipv6", "uuid") @@ -2919,28 +2922,24 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): web_search_results: Sequence[object], completion_response: Mapping[str, object], ) -> list[object]: - content: Final = completion_response.get("content") - blocks: Final = content if isinstance(content, Sequence) else () - inputs: Final = { - call_id: tool_input - for block in blocks - if isinstance(block, Mapping) - and block.get("type") == "server_tool_use" - and block.get("name") == "web_search" - and isinstance((call_id := block.get("id")), str) - and isinstance((tool_input := block.get("input")), Mapping) + blocks: Final = _RESPONSE_BLOCKS.validate_python(completion_response.get("content") or ()) + results_by_id: Final = { + tool_use_id: result + for result in _RESPONSE_BLOCKS.validate_python(web_search_results) + if result.get("type") == "web_search_tool_result" + and isinstance((tool_use_id := result.get("tool_use_id")), str) } return [ build_web_search_call( - tool_id=tool_use_id, - tool_input=inputs.get(tool_use_id, {}), - result=result, + tool_id=call_id, + tool_input=block.get("input"), + result=results_by_id.get(call_id, _PENDING_WEB_SEARCH_RESULT), + status=None if call_id in results_by_id else "in_progress", ) - for result in web_search_results - if isinstance(result, dict) - and result.get("type") == "web_search_tool_result" - and isinstance((tool_use_id := result.get("tool_use_id")), str) - and tool_use_id in inputs + for block in blocks + if block.get("type") == "server_tool_use" + and block.get("name") == "web_search" + and isinstance((call_id := block.get("id")), str) ] def _build_provider_specific_fields( @@ -2964,10 +2963,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): if web_search_results is not None: provider_specific_fields["web_search_results"] = web_search_results - provider_specific_fields["web_search_calls"] = self._build_web_search_calls( - web_search_results, - completion_response, - ) + + web_search_calls: Final = self._build_web_search_calls(web_search_results or (), completion_response) + if web_search_results is not None or web_search_calls: + provider_specific_fields["web_search_calls"] = web_search_calls if tool_results is not None: provider_specific_fields["tool_results"] = tool_results @@ -2989,10 +2988,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): if compaction_blocks is not None: provider_specific_fields["compaction_blocks"] = compaction_blocks - refusal_fields: Final = refusal_provider_specific_fields( - REFUSAL_STOP_FIELDS.validate_python(completion_response) + stop_reason_fields: Final = stop_reason_provider_specific_fields( + STOP_REASON_FIELDS.validate_python(completion_response) ) - return {**provider_specific_fields, **refusal_fields} + return {**provider_specific_fields, **stop_reason_fields} def transform_parsed_response( self, @@ -3067,22 +3066,26 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): if json_extra_content: merged_text = merged_text + json_extra_content if merged_text else json_extra_content + refusal_fields: Final = refusal_params(STOP_REASON_FIELDS.validate_python(completion_response)) _message = litellm.Message( tool_calls=tool_calls_for_message, content=merged_text or None, provider_specific_fields=provider_specific_fields, thinking_blocks=thinking_blocks, reasoning_content=reasoning_content, + **refusal_fields, ) _message.provider_specific_fields = provider_specific_fields if json_mode_message is not None: - refusal_fields: Final = refusal_provider_specific_fields( - REFUSAL_STOP_FIELDS.validate_python(completion_response) + stop_reason_fields: Final = stop_reason_provider_specific_fields( + STOP_REASON_FIELDS.validate_python(completion_response) ) completion_response["stop_reason"] = "stop" - if refusal_fields: - json_mode_message.provider_specific_fields = dict(refusal_fields) + if stop_reason_fields: + json_mode_message.provider_specific_fields = dict(stop_reason_fields) + if "refusal" in refusal_fields: + json_mode_message.refusal = refusal_fields["refusal"] if thinking_blocks: json_mode_message.thinking_blocks = thinking_blocks json_mode_message.reasoning_content = reasoning_content diff --git a/litellm/llms/anthropic/common_utils.py b/litellm/llms/anthropic/common_utils.py index e0da8943a4a..4f5dc678bae 100644 --- a/litellm/llms/anthropic/common_utils.py +++ b/litellm/llms/anthropic/common_utils.py @@ -2518,7 +2518,7 @@ def create_anthropic_model_list_response( } -class RefusalStopFields(TypedDict, total=False): +class StopReasonFields(TypedDict, total=False): stop_reason: ReadOnly[object] stop_details: ReadOnly[object] @@ -2535,16 +2535,38 @@ def _dict_or_empty(value: object) -> object: return value if _is_dict(value) else _no_stop_fields() -REFUSAL_STOP_FIELDS: Final[TypeAdapter[RefusalStopFields]] = TypeAdapter( - Annotated[RefusalStopFields, BeforeValidator(_dict_or_empty)] +STOP_REASON_FIELDS: Final[TypeAdapter[StopReasonFields]] = TypeAdapter( + Annotated[StopReasonFields, BeforeValidator(_dict_or_empty)] ) +_STOP_REASONS_KEPT_RAW: Final = frozenset({"refusal", "pause_turn"}) -def refusal_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]: + +def stop_reason_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]: stop_reason: Final = message.get("stop_reason") stop_details: Final = message.get("stop_details") if stop_details is not None: return {"stop_reason": stop_reason, "stop_details": stop_details} - if stop_reason == "refusal": + if stop_reason in _STOP_REASONS_KEPT_RAW: return {"stop_reason": stop_reason} return {} + + +class _RefusalStopDetails(TypedDict, total=False): + explanation: ReadOnly[object] + + +_REFUSAL_STOP_DETAILS: Final[TypeAdapter[_RefusalStopDetails]] = TypeAdapter( + Annotated[_RefusalStopDetails, BeforeValidator(_dict_or_empty)] +) + + +class RefusalParams(TypedDict, total=False): + refusal: ReadOnly[str] + + +def refusal_params(message: Mapping[str, object]) -> RefusalParams: + if message.get("stop_reason") != "refusal": + return {} + explanation: Final = _REFUSAL_STOP_DETAILS.validate_python(message.get("stop_details")).get("explanation") + return {"refusal": explanation} if isinstance(explanation, str) and explanation else {} diff --git a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py index b19c0d8e2a9..61d23c2de0d 100644 --- a/litellm/llms/anthropic/messages/adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/messages/adapters/streaming_iterator.py @@ -33,6 +33,7 @@ from litellm.llms.anthropic.messages.safeguards import StreamedSafeguardResults from litellm.llms.anthropic.messages.utils import anthropic_sse_frame from litellm.llms.base_llm.chat.transformation import BaseLLMException from litellm.types.llms.anthropic import ( + AnthropicStopDetails, AppliedEdit, CompactionBlock, ContentBlockDelta, @@ -284,9 +285,7 @@ class _CombinedChunkSplitter: @staticmethod def _is_combined(chunk: "ModelResponseStream") -> bool: """True if ``chunk`` carries response content AND a finish_reason.""" - from litellm.llms.anthropic.messages.native.utils import ( - openai_chat_refusal_text, - ) + from litellm.llms.anthropic.messages.native.utils import refusal_reply_text choices: Final = _optional_attr_sequence(chunk, "choices") if not choices: @@ -302,7 +301,7 @@ class _CombinedChunkSplitter: or _optional_attr(delta, "tool_calls") or _optional_attr(delta, "reasoning_content") or _optional_attr(delta, "thinking_blocks") - or openai_chat_refusal_text(delta) + or refusal_reply_text(delta) ) _PAYLOAD_FIELD_GROUPS: "tuple[tuple[str, ...], ...]" = ( @@ -659,6 +658,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): self.compaction_block = compaction_block self.iterations_usage = iterations_usage self._refusal_text: str = "" + self._paused_turn: bool = False + self._claude_refusal_stop_details: AnthropicStopDetails | None = None self.sent_compaction_block: bool = False # Per-phase flags so the compaction block's start/delta/stop events # are emitted (and the public state machine is advanced) in @@ -953,7 +954,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): current_content_block_index=self.current_content_block_index, applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None), ) - processed_chunk = self._with_refusal_stop_details(processed_chunk) + processed_chunk = self._with_stop_reason_fields(processed_chunk) # Check if this is a usage chunk and we have a held stop_reason chunk if will_merge_into_held: @@ -1188,7 +1189,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): current_content_block_index=self.current_content_block_index, applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None), ) - processed_chunk = self._with_refusal_stop_details(processed_chunk) + processed_chunk = self._with_stop_reason_fields(processed_chunk) # Check if this is a usage chunk and we have a held stop_reason chunk if will_merge_into_held: @@ -1406,14 +1407,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): def _increment_content_block_index(self): self.current_content_block_index += 1 - def _with_refusal_stop_details( + def _with_stop_reason_fields( self, processed_chunk: ContentBlockDelta | MessageBlockDelta, ) -> ContentBlockDelta | MessageBlockDelta: - if processed_chunk["type"] != "message_delta" or not self._refusal_text: + if processed_chunk["type"] != "message_delta": return processed_chunk delta: Final = processed_chunk["delta"] - if delta.get("stop_reason") == "max_tokens": + if self._paused_turn: + paused_delta: Final[MessageDelta] = {**delta, "stop_reason": "pause_turn"} + paused_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": paused_delta} + return paused_chunk + if delta.get("stop_reason") == "refusal" and self._claude_refusal_stop_details is not None: + claude_refusal_delta: Final[MessageDelta] = {**delta, "stop_details": self._claude_refusal_stop_details} + claude_refusal_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": claude_refusal_delta} + return claude_refusal_chunk + if not self._refusal_text or delta.get("stop_reason") == "max_tokens": return processed_chunk from litellm.llms.anthropic.messages.native.utils import ( refusal_stop_details, @@ -1470,9 +1479,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): @staticmethod def _is_blank_delta(chunk: "ModelResponseStream") -> bool: from litellm.llms.anthropic.common_utils import is_empty_unsigned_thinking_block - from litellm.llms.anthropic.messages.native.utils import ( - openai_chat_refusal_text, - ) + from litellm.llms.anthropic.messages.native.utils import refusal_reply_text choice: Final = chunk.choices[0] if choice.finish_reason is not None: @@ -1482,7 +1489,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): return False if getattr(delta, "content", None): return False - if openai_chat_refusal_text(delta): + if refusal_reply_text(delta): return False if getattr(delta, "reasoning_content", None): return False @@ -1497,6 +1504,18 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): return False return True + def _record_anthropic_stop_fields(self, delta: object) -> None: + from litellm.llms.anthropic.messages.native.utils import ( + anthropic_refusal_stop_details, + anthropic_stop_reason, + ) + + if anthropic_stop_reason(delta) == "pause_turn": + self._paused_turn = True + claude_refusal_stop_details: Final = anthropic_refusal_stop_details(delta) + if claude_refusal_stop_details is not None: + self._claude_refusal_stop_details = claude_refusal_stop_details + def _should_start_new_content_block(self, chunk: "ModelResponseStream") -> bool: """ Determine if we should start a new content block based on the processed chunk. @@ -1507,16 +1526,15 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper): - Different content types in the response - Specific markers in the content """ - from litellm.llms.anthropic.messages.native.utils import ( - openai_chat_refusal_text, - ) + from litellm.llms.anthropic.messages.native.utils import refusal_reply_text from .transformation import LiteLLMAnthropicMessagesAdapter + self._record_anthropic_stop_fields(chunk.choices[0].delta) if chunk.choices[0].finish_reason is not None: return False - refusal_text: Final = openai_chat_refusal_text(chunk.choices[0].delta) + refusal_text: Final = refusal_reply_text(chunk.choices[0].delta) if refusal_text is not None: self._refusal_text = self._refusal_text + refusal_text diff --git a/litellm/llms/anthropic/messages/adapters/transformation.py b/litellm/llms/anthropic/messages/adapters/transformation.py index bc9649ed867..0e4eb29ac0f 100644 --- a/litellm/llms/anthropic/messages/adapters/transformation.py +++ b/litellm/llms/anthropic/messages/adapters/transformation.py @@ -150,7 +150,9 @@ from litellm.llms.anthropic.messages.native.mid_conversation_system import ( is_system_role_message, ) from litellm.llms.anthropic.messages.native.utils import ( - openai_chat_refusal_text, + anthropic_refusal_stop_details, + anthropic_stop_reason, + refusal_reply_text, refusal_stop_details, ) from litellm.types.llms.anthropic import ( @@ -1467,7 +1469,7 @@ class LiteLLMAnthropicMessagesAdapter: new_content.append( AnthropicResponseContentBlockText(type="text", text=choice.message.content).model_dump() ) - if (refusal_text := openai_chat_refusal_text(choice.message)) is not None: + if (refusal_text := refusal_reply_text(choice.message)) is not None: new_content.append(AnthropicResponseContentBlockText(type="text", text=refusal_text).model_dump()) # Handle tool calls (in parallel to text content) if choice.message.tool_calls is not None and len(choice.message.tool_calls) > 0: @@ -1634,7 +1636,15 @@ class LiteLLMAnthropicMessagesAdapter: tool_name_mapping=tool_name_mapping, ) refusal_text: Final = next( - (text for choice in response.choices if (text := openai_chat_refusal_text(choice.message)) is not None), + (text for choice in response.choices if (text := refusal_reply_text(choice.message)) is not None), + None, + ) + claude_refusal_stop_details: Final = next( + ( + details + for choice in response.choices + if (details := anthropic_refusal_stop_details(choice.message)) is not None + ), None, ) @@ -1643,8 +1653,10 @@ class LiteLLMAnthropicMessagesAdapter: ## extract finish reason openai_finish_reason: Final = response.choices[0].finish_reason if response.choices else "stop" - translated_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic( - openai_finish_reason=openai_finish_reason + translated_finish_reason: Final = ( + "pause_turn" + if response.choices and anthropic_stop_reason(response.choices[0].message) == "pause_turn" + else self._translate_openai_finish_reason_to_anthropic(openai_finish_reason=openai_finish_reason) ) anthropic_finish_reason: Final = ( "compaction" @@ -1686,7 +1698,11 @@ class LiteLLMAnthropicMessagesAdapter: usage=anthropic_usage, content=anthropic_content, stop_reason=anthropic_finish_reason, - stop_details=(refusal_stop_details(refusal_text) if anthropic_finish_reason == "refusal" else None), + stop_details=( + (claude_refusal_stop_details or refusal_stop_details(refusal_text)) + if anthropic_finish_reason == "refusal" + else None + ), ) applied_edits: Final = polyfill_result.applied_edits_for_response() if polyfill_result else None @@ -1727,7 +1743,7 @@ class LiteLLMAnthropicMessagesAdapter: "signature": thought_sig, } return "tool_use", cast("ContentBlockContentBlockDict", tool_block) - elif (choice.delta.content is not None and len(choice.delta.content) > 0) or openai_chat_refusal_text( + elif (choice.delta.content is not None and len(choice.delta.content) > 0) or refusal_reply_text( choice.delta ) is not None: return "text", TextBlock(type="text", text="") @@ -1810,7 +1826,7 @@ class LiteLLMAnthropicMessagesAdapter: return "thinking_delta", ContentThinkingBlockDelta(type="thinking_delta", thinking=reasoning_content) else: refusal_text: Final = "".join( - refusal for choice in choices if (refusal := openai_chat_refusal_text(choice.delta)) is not None + refusal for choice in choices if (refusal := refusal_reply_text(choice.delta)) is not None ) return "text_delta", ContentTextBlockDelta(type="text_delta", text=text + refusal_text) diff --git a/litellm/llms/anthropic/messages/native/utils.py b/litellm/llms/anthropic/messages/native/utils.py index 29bbd8178ba..ea3afdb9a8f 100644 --- a/litellm/llms/anthropic/messages/native/utils.py +++ b/litellm/llms/anthropic/messages/native/utils.py @@ -2,7 +2,7 @@ from collections.abc import Iterable, Mapping, Sequence from functools import lru_cache from typing import TYPE_CHECKING, Final, cast, get_type_hints -from pydantic import JsonValue, TypeAdapter, ValidationError +from pydantic import ConfigDict, JsonValue, TypeAdapter, ValidationError, with_config from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook from litellm.litellm_core_utils.core_helpers import as_str_mapping @@ -162,6 +162,34 @@ def openai_chat_refusal_text(message_or_delta: object) -> str | None: return _mapping_str_field(getattr(message_or_delta, "provider_specific_fields", None), "refusal") +def anthropic_stop_reason(message_or_delta: object) -> str | None: + return _mapping_str_field(getattr(message_or_delta, "provider_specific_fields", None), "stop_reason") + + +def refusal_reply_text(message_or_delta: object) -> str | None: + if anthropic_stop_reason(message_or_delta) == "refusal": + return None + return openai_chat_refusal_text(message_or_delta) + + +@with_config(ConfigDict(extra="allow")) +class _AnthropicRefusalStopDetails(AnthropicStopDetails, total=False): + pass + + +_ANTHROPIC_REFUSAL_STOP_DETAILS: Final = TypeAdapter(_AnthropicRefusalStopDetails) + + +def anthropic_refusal_stop_details(message_or_delta: object) -> AnthropicStopDetails | None: + provider_specific_fields: Final = getattr(message_or_delta, "provider_specific_fields", None) + if _mapping_str_field(provider_specific_fields, "stop_reason") != "refusal": + return None + try: + return _ANTHROPIC_REFUSAL_STOP_DETAILS.validate_python(_mapping_field(provider_specific_fields, "stop_details")) + except ValidationError: + return refusal_stop_details(openai_chat_refusal_text(message_or_delta)) + + def _responses_message_refusal_text(item: object) -> str | None: from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py index 559daf767c2..ddda9e7fb46 100644 --- a/litellm/llms/openai/responses/guardrail_translation/handler.py +++ b/litellm/llms/openai/responses/guardrail_translation/handler.py @@ -38,6 +38,7 @@ from itertools import accumulate, chain, groupby, repeat from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeVar, Union, cast +from openai.types.responses.response_output_refusal import ResponseOutputRefusal from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError from typing_extensions import ReadOnly, TypedDict @@ -277,6 +278,7 @@ _TOOL_CALL_PAYLOAD_EVENT_TYPES: Final = _TOOL_CALL_PAYLOAD_DELTA_EVENT_TYPES | f ) _OUTPUT_ITEM_EVENT_TYPES: Final = frozenset({"response.output_item.added", "response.output_item.done"}) _OUTPUT_TEXT_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.output_text.done"}) +_TEXT_DELTA_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.refusal.delta"}) _TOOL_OUTPUT_TEXT_PART_TYPES: Final = frozenset({"input_text", "output_text", "text"}) _PATCHABLE_ITEM_FIELDS: Final[Mapping[str, str]] = MappingProxyType( {"function_call_output": "output", "custom_tool_call_output": "output", "message": "content"} @@ -537,6 +539,14 @@ def _scanned_text_field(item: Mapping[str, object]) -> str: return "output" if item.get("type") in TOOL_CALL_OUTPUT_ITEM_TYPES else "content" +def _content_part_text_field(part: object) -> str: + return "refusal" if stream_item_field(part, "type") == "refusal" else "text" + + +def _has_writable_field(part: object, field: str) -> bool: + return isinstance(part, dict) or hasattr(part, field) + + def _tool_output_part_text(part: object) -> str | None: if not isinstance(part, dict): return None @@ -993,7 +1003,7 @@ class OpenAIResponsesHandler(BaseTranslation): for content_idx, content_item in enumerate(content): if isinstance(content_item, dict): # Extract text - text_str = content_item.get("text", None) + text_str = content_item.get(_content_part_text_field(content_item), None) if text_str is not None: texts_to_check.append(text_str) task_mappings.append((msg_idx, int(content_idx))) @@ -1042,8 +1052,7 @@ class OpenAIResponsesHandler(BaseTranslation): elif isinstance(content, list): # Replace specific text item in list content - if isinstance(messages[msg_idx][field][content_idx_optional], dict): - messages[msg_idx][field][content_idx_optional]["text"] = guardrail_response + self._write_content_part_text(content[content_idx_optional], guardrail_response) async def process_output_response( self, @@ -1380,7 +1389,8 @@ class OpenAIResponsesHandler(BaseTranslation): if isinstance(output_index, int) and isinstance(content_index, int) ) if len(placeable_positions) != len(scanned_positions) or any( - stream_item_field(event, "type") not in _OUTPUT_TEXT_EVENT_TYPES for event in scanned_events + stream_item_field(event, "type") not in _OUTPUT_TEXT_EVENT_TYPES | _TEXT_DELTA_EVENT_TYPES + for event in scanned_events ): from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite @@ -1401,6 +1411,12 @@ class OpenAIResponsesHandler(BaseTranslation): else: setattr(event, field, value) + @staticmethod + def _write_content_part_text(part: object, text: str) -> None: + field: Final = _content_part_text_field(part) + if _has_writable_field(part, field): + OpenAIResponsesHandler._write_event_field(part, field, text) + @staticmethod def _write_compaction_summary(item: object, summary: str) -> None: encrypted_content: Final = stream_item_field(item, "encrypted_content") @@ -1416,11 +1432,12 @@ class OpenAIResponsesHandler(BaseTranslation): ) -> None: """Sync pre-completion stream events with the rewritten completed response, keyed by ``(output_index, content_index)``: the first - ``output_text.delta`` for a rewritten item carries the full rewritten - text and the rest are blanked, while ``output_text.done``, - ``content_part.done``, and ``output_item.done`` events carry the full - rewritten text, so every event a client may read agrees with the - rewritten ``response.completed`` payload.""" + ``output_text.delta`` or ``refusal.delta`` for a rewritten item carries + the full rewritten text and the rest are blanked, while + ``output_text.done``, ``refusal.done``, ``content_part.done``, and + ``output_item.done`` events carry the full rewritten text, so every + event a client may read agrees with the rewritten ``response.completed`` + payload.""" delta_replacements: Final = MappingProxyType( {position: chain((rewritten,), repeat("")) for position, rewritten in rewrites_by_position.items()} ) @@ -1436,14 +1453,14 @@ class OpenAIResponsesHandler(BaseTranslation): if not isinstance(output_index, int) or not isinstance(content_index, int): continue position = (output_index, content_index) - if event_type == "response.output_text.delta" and position in delta_replacements: + if event_type in _TEXT_DELTA_EVENT_TYPES and position in delta_replacements: self._write_event_field(event, "delta", next(delta_replacements[position])) elif event_type == "response.output_text.done" and position in rewrites_by_position: self._write_event_field(event, "text", rewrites_by_position[position]) + elif event_type == "response.refusal.done" and position in rewrites_by_position: + self._write_event_field(event, "refusal", rewrites_by_position[position]) elif event_type == "response.content_part.done" and position in rewrites_by_position: - part = event.get("part") - if isinstance(part, dict) or hasattr(part, "text"): - self._write_event_field(part, "text", rewrites_by_position[position]) + self._write_content_part_text(event.get("part"), rewrites_by_position[position]) @staticmethod def _sync_output_item_done_event( @@ -1459,7 +1476,7 @@ class OpenAIResponsesHandler(BaseTranslation): for (item_idx, content_idx), rewritten in rewrites_by_position.items(): if item_idx != output_index or not 0 <= content_idx < len(content): continue - OpenAIResponsesHandler._write_event_field(content[content_idx], "text", rewritten) + OpenAIResponsesHandler._write_content_part_text(content[content_idx], rewritten) def _deliver_ended_stream_tool_call_rewrites( self, @@ -1814,7 +1831,7 @@ class OpenAIResponsesHandler(BaseTranslation): return # Handle both GenericResponseOutputItem and dict - content: list[OutputText] | list[dict] | None = None + content: list[OutputText | ResponseOutputRefusal] | list[dict[str, object]] | None = None if isinstance(output_item, BaseModel): try: output_item_dump: Final = output_item.model_dump() @@ -1837,17 +1854,9 @@ class OpenAIResponsesHandler(BaseTranslation): verbose_proxy_logger.debug("OpenAI Responses API: Processing output item: %s", output_item) - # Iterate through content items (list of OutputText objects) for content_idx, content_item in enumerate(content): - # Handle both OutputText objects and dicts - if isinstance(content_item, OutputText): - text_content = content_item.text - elif isinstance(content_item, dict): - text_content = content_item.get("text") - else: - continue - - if text_content: + text_content = stream_item_field(content_item, _content_part_text_field(content_item)) + if isinstance(text_content, str) and text_content: texts_to_check.append(text_content) task_mappings.append((output_idx, int(content_idx))) @@ -1887,33 +1896,24 @@ class OpenAIResponsesHandler(BaseTranslation): # Handle both GenericResponseOutputItem, BaseModel, and dict if isinstance(output_item, GenericResponseOutputItem): if output_item.content and content_idx < len(output_item.content): - content_item = output_item.content[content_idx] - if isinstance(content_item, OutputText): - content_item.text = guardrail_response - elif isinstance(content_item, dict): - content_item["text"] = guardrail_response + self._write_content_part_text(output_item.content[content_idx], guardrail_response) elif isinstance(output_item, BaseModel): # Handle other Pydantic models by converting to GenericResponseOutputItem try: generic_item = GenericResponseOutputItem.model_validate(output_item.model_dump()) - if generic_item.content and content_idx < len(generic_item.content): - content_item = generic_item.content[content_idx] - if isinstance(content_item, OutputText): - content_item.text = guardrail_response - # Update the original response output - if hasattr(output_item, "content") and output_item.content: - original_content = output_item.content[content_idx] - if hasattr(original_content, "text"): - original_content.text = guardrail_response + if ( + generic_item.content + and content_idx < len(generic_item.content) + and hasattr(output_item, "content") + and output_item.content + ): + self._write_content_part_text(output_item.content[content_idx], guardrail_response) except Exception: pass elif isinstance(output_item, dict): content = output_item.get("content", []) if content and content_idx < len(content): - if isinstance(content[content_idx], dict): - content[content_idx]["text"] = guardrail_response - elif hasattr(content[content_idx], "text"): - content[content_idx].text = guardrail_response + self._write_content_part_text(content[content_idx], guardrail_response) def build_block_sse_chunks( self, diff --git a/litellm/proxy/_lazy_openapi_snapshot.json b/litellm/proxy/_lazy_openapi_snapshot.json index 41ab7fcaf88..546022bc0c4 100644 --- a/litellm/proxy/_lazy_openapi_snapshot.json +++ b/litellm/proxy/_lazy_openapi_snapshot.json @@ -24268,7 +24268,14 @@ "properties": { "content": { "items": { - "$ref": "#/components/schemas/OutputText" + "anyOf": [ + { + "$ref": "#/components/schemas/OutputText" + }, + { + "$ref": "#/components/schemas/ResponseOutputRefusal" + } + ] }, "title": "Content", "type": "array" diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index 892da541935..c39a53c782b 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -9,6 +9,7 @@ import litellm from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items from litellm.litellm_core_utils.hidden_params import get_or_create_hidden_params, served_from_cache from litellm.litellm_core_utils.litellm_logging import BilledRoundsSpend +from litellm.llms.anthropic.messages.native.utils import openai_chat_refusal_text from litellm.main import stream_chunk_builder from litellm.responses.litellm_completion_transformation.compaction_items import compaction_output_item from litellm.responses.litellm_completion_transformation.custom_tools import ( @@ -35,6 +36,7 @@ from litellm.types.llms.openai import ( ContentPartAddedEvent, ContentPartDoneEvent, ContentPartDonePartOutputText, + ContentPartDonePartRefusal, FunctionCallArgumentsDeltaEvent, FunctionCallArgumentsDoneEvent, OutputItemAddedEvent, @@ -45,6 +47,8 @@ from litellm.types.llms.openai import ( ReasoningSummaryPartDoneEvent, ReasoningSummaryTextDeltaEvent, ReasoningSummaryTextDoneEvent, + RefusalDeltaEvent, + RefusalDoneEvent, ResponseCompletedEvent, ResponseCreatedEvent, ResponseIncompleteEvent, @@ -110,6 +114,7 @@ def _delta_carries_output(delta: ChatCompletionDelta) -> bool: delta.function_call, getattr(delta, "images", None), getattr(delta, "audio", None), + openai_chat_refusal_text(delta), ) return any(fields) or _delta_has_signed_thinking_block(delta) @@ -119,6 +124,13 @@ def _delta_carries_reasoning_and_answer(delta: ChatCompletionDelta) -> bool: return has_reasoning and bool(delta.content or delta.tool_calls) +_WEB_SEARCH_PROGRESS_EVENTS: Final = ( + (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS, WebSearchCallInProgressEvent), + (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING, WebSearchCallSearchingEvent), + (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED, WebSearchCallCompletedEvent), +) + + class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): """ Async iterator for processing streaming responses from the Responses API. @@ -152,6 +164,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): self.sent_output_text_done_event: bool = False self.sent_output_content_part_done_event: bool = False self.sent_output_item_done_event: bool = False + self._sent_refusal_events: bool = False self._annotation_index: int = 0 self._annotations_awaiting_message: tuple[ChatCompletionAnnotation, ...] = () self.litellm_model_response: ModelResponse | TextCompletionResponse | None = None @@ -592,11 +605,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): } ), ) - for event_type, event_class in ( - (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS, WebSearchCallInProgressEvent), - (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING, WebSearchCallSearchingEvent), - (ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED, WebSearchCallCompletedEvent), - ): + progress_events: Final = ( + _WEB_SEARCH_PROGRESS_EVENTS[:1] if item.status == "in_progress" else _WEB_SEARCH_PROGRESS_EVENTS + ) + for event_type, event_class in progress_events: self._sequence_number += 1 event = event_class(type=event_type, output_index=output_index, item_id=item.id) event.__dict__["sequence_number"] = self._sequence_number @@ -798,9 +810,14 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): return event def _queue_message_item_added_events(self) -> None: + if not self.sent_message_item_added_event: + self._queue_message_item_added_event() + self.sent_content_part_added_event = True + self._pending_response_events.append(self.create_content_part_added_event()) + + def _queue_message_item_added_event(self) -> None: item_id: Final = self._message_item_id() self.sent_message_item_added_event = True - self.sent_content_part_added_event = True self._sequence_number += 1 event: Final = OutputItemAddedEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, @@ -817,7 +834,47 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): ) event.__dict__["sequence_number"] = self._sequence_number self._pending_response_events.append(event) - self._pending_response_events.append(self.create_content_part_added_event()) + + def _queue_refusal_events(self, refusal: str) -> None: + item_id: Final = self._message_item_id() + output_index: Final = self._message_index() + content_index: Final = 1 if self.sent_content_part_added_event else 0 + events: Final = ( + ContentPartAddedEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + item_id=item_id, + output_index=output_index, + content_index=content_index, + part=ContentPartDonePartRefusal(type="refusal", refusal=""), + ), + RefusalDeltaEvent( + type=ResponsesAPIStreamEvents.REFUSAL_DELTA, + item_id=item_id, + output_index=output_index, + content_index=content_index, + delta=refusal, + ), + RefusalDoneEvent( + type=ResponsesAPIStreamEvents.REFUSAL_DONE, + item_id=item_id, + output_index=output_index, + content_index=content_index, + refusal=refusal, + ), + ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + item_id=item_id, + output_index=output_index, + content_index=content_index, + part=ContentPartDonePartRefusal(type="refusal", refusal=refusal), + ), + ) + first_sequence_number: Final = self._sequence_number + 1 + self._sequence_number += len(events) + self._pending_response_events.extend( + event.model_copy(update={"sequence_number": sequence_number}) + for sequence_number, event in enumerate(events, start=first_sequence_number) + ) def _merge_provider_specific_fields(self, src: dict) -> None: """Merge provider_specific_fields using last-value-wins for lists. @@ -974,14 +1031,22 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): if self._cached_item_id is None: self._cached_item_id = f"msg_{uuid.uuid4()}" - text: Final = litellm_complete_object.choices[0].message.content or "" - annotations = getattr(litellm_complete_object.choices[0].message, "annotations", None) + message: Final = litellm_complete_object.choices[0].message + text: Final = message.content or "" + annotations = getattr(message, "annotations", None) response_annotations: Final = ( LiteLLMCompletionResponsesConfig.transform_chat_completion_annotations_to_response_output_annotations( annotations=annotations ) ) + text_parts: Final = ( + ({"type": "output_text", "text": text, "annotations": response_annotations},) + if self.sent_content_part_added_event + else () + ) + refusal: Final = openai_chat_refusal_text(message) + refusal_parts: Final = ({"type": "refusal", "refusal": refusal},) if refusal is not None else () return OutputItemDoneEvent( type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=self._message_index(), @@ -992,13 +1057,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): "status": "completed", "type": "message", "role": "assistant", - "content": [ - { - "type": "output_text", - "text": text, - "annotations": response_annotations, - } - ], + "content": [*text_parts, *refusal_parts], } ), ) @@ -1050,22 +1109,33 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): def return_default_done_events( self, litellm_complete_object: ModelResponse ) -> BaseLiteLLMOpenAIResponseObject | None: + message: Final = litellm_complete_object.choices[0].message + refusal: Final = openai_chat_refusal_text(message) if self.sent_message_item_added_event is False: - final_content: Final = litellm_complete_object.choices[0].message.content or "" - if not final_content: + final_content: Final = message.content or "" + if not final_content and refusal is None: self.sent_output_text_done_event = True self.sent_output_content_part_done_event = True self.sent_output_item_done_event = True return None - self._queue_message_item_added_events() + if final_content: + self._queue_message_item_added_events() + else: + self._queue_message_item_added_event() return self._pending_response_events.pop(0) - if self.sent_output_text_done_event is False: + if self.sent_output_text_done_event is False and self.sent_content_part_added_event: self.sent_output_text_done_event = True return self.create_output_text_done_event(litellm_complete_object) - if self.sent_output_content_part_done_event is False: + if self.sent_output_content_part_done_event is False and self.sent_content_part_added_event: self.sent_output_content_part_done_event = True return self.create_output_content_part_done_event(litellm_complete_object) + if refusal is not None and not self._sent_refusal_events: + self._sent_refusal_events = True + self._queue_refusal_events(refusal) + return self._pending_response_events.pop(0) if self.sent_output_item_done_event is False: + self.sent_output_text_done_event = True + self.sent_output_content_part_done_event = True self.sent_output_item_done_event = True return self.create_output_item_done_event(litellm_complete_object) return None @@ -1169,7 +1239,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): # DO NOT create message item return - # Default: message + if openai_chat_refusal_text(delta) is not None and not delta.content: + self._queue_message_item_added_event() + return self._queue_message_item_added_events() return @@ -1413,7 +1485,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): # Priority 2: Handle text deltas delta_content: Final = self._get_delta_string_from_streaming_choices(chunk.choices) if delta_content: - if not self.sent_message_item_added_event: + if not self.sent_content_part_added_event: self._queue_message_item_added_events() self._hold_tool_calls_behind_text(getattr(chunk.choices[0].delta, "tool_calls", None)) self._sequence_number += 1 diff --git a/litellm/responses/litellm_completion_transformation/transformation.py b/litellm/responses/litellm_completion_transformation/transformation.py index 2e156d5abba..65e3b301676 100644 --- a/litellm/responses/litellm_completion_transformation/transformation.py +++ b/litellm/responses/litellm_completion_transformation/transformation.py @@ -8,7 +8,7 @@ import uuid from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence from dataclasses import dataclass from functools import partial -from itertools import chain, groupby +from itertools import accumulate, chain, groupby from types import MappingProxyType from typing import ( TYPE_CHECKING, @@ -28,7 +28,7 @@ from openai.types.chat.chat_completion_named_tool_choice_param import ( from openai.types.chat.chat_completion_named_tool_choice_param import ( Function as NamedToolChoiceFunction, ) -from openai.types.responses import ResponseFunctionToolCall, ResponseFunctionWebSearch +from openai.types.responses import ResponseFunctionToolCall, ResponseFunctionWebSearch, ResponseOutputRefusal from openai.types.responses.response_compaction_item import ResponseCompactionItem from openai.types.responses.response_create_params import ResponseInputParam from openai.types.responses.tool_choice_custom_param import ToolChoiceCustomParam @@ -65,6 +65,7 @@ from litellm.litellm_core_utils.hidden_params import ( ) from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control +from litellm.llms.anthropic.messages.native.utils import anthropic_stop_reason, openai_chat_refusal_text from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM from litellm.llms.bedrock.common_utils import bedrock_declared_cache_point_support from litellm.llms.openrouter.chat.transformation import openrouter_model_supports_cache_control @@ -446,6 +447,32 @@ def _input_item_role(input_item: Mapping[str, object]) -> str: return cast(str, input_item.get("role") or "user") # cast-ok: client-supplied role forwarded verbatim, unvalidated +def _starts_a_turn(input_item: Mapping[str, object]) -> bool: + item_type: Final = input_item.get("type") + if item_type in TOOL_CALL_OUTPUT_ITEM_TYPES: + return True + return item_type in (None, "message") and input_item.get("role") != "assistant" + + +def _is_unrun_hosted_search(input_item: Mapping[str, object]) -> bool: + return input_item.get("type") == "web_search_call" and input_item.get("status") == "in_progress" + + +def _without_encrypted_content(input_item: Mapping[str, object]) -> Mapping[str, object]: + return {key: value for key, value in input_item.items() if key != "encrypted_content"} + + +def _without_paused_turn_thinking(input_items: Sequence[Mapping[str, object]]) -> tuple[Mapping[str, object], ...]: + turns: Final = tuple(accumulate(int(_starts_a_turn(item)) for item in input_items)) + paused_turns: Final = frozenset( + turn for turn, item in zip(turns, input_items, strict=True) if _is_unrun_hosted_search(item) + ) + return tuple( + _without_encrypted_content(item) if turn in paused_turns and item.get("type") == "reasoning" else item + for turn, item in zip(turns, input_items, strict=True) + ) + + SessionMessage: TypeAlias = ( AllMessageValues | GenericChatCompletionMessage @@ -1416,7 +1443,7 @@ class LiteLLMCompletionResponsesConfig: messages.append(ChatCompletionUserMessage(role="user", content=input)) elif isinstance(input, list): existing_tool_call_ids: Final[set[str]] = set() - for _input in input: + for _input in _without_paused_turn_thinking(input): chat_completion_messages = ( LiteLLMCompletionResponsesConfig._transform_responses_api_input_item_to_chat_completion_message( input_item=_input, @@ -2528,45 +2555,55 @@ class LiteLLMCompletionResponsesConfig: for item in content: if isinstance(item, str): content_list.append(item) - elif isinstance(item, dict): - if ( - media := decode_responses_media( - fields := _validate_responses_media(_STR_KEY_DICT_ADAPTER, item) + elif ( + isinstance(item, dict) + and ( + part + := LiteLLMCompletionResponsesConfig._transform_responses_api_content_part_to_chat_completion_part( + _validate_responses_media(_STR_KEY_DICT_ADAPTER, item) ) - ) is not None: - content_list.append(dict(media)) - elif fields.get("type") == "input_image": - content_list.append(dict(_responses_image_part(fields))) - elif fields.get("type") == "input_file": - content_list.append(dict(_responses_file_part(fields))) - elif fields.get("type") == "input_audio": - content_list.append(dict(_validate_responses_media(_AUDIO_PART_ADAPTER, fields))) - elif fields.get("type") == "encrypted_content": - encrypted_content = fields.get("encrypted_content") - if encrypted_content is not None: - content_list.append( - OpenAIChatCompletionTextObject(type="text", text=str(encrypted_content)) - ) - else: - # Skip text blocks with None text to avoid downstream errors - text_value = fields.get("text") - if text_value is None: - continue - content_block: dict[str, object] = { - "type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type( - content_type - if isinstance(content_type := fields.get("type"), str) and content_type - else "text" - ), - "text": text_value, - } - if "cache_control" in fields: - content_block["cache_control"] = fields["cache_control"] - content_list.append(content_block) + ) + is not None + ): + content_list.append(part) return content_list else: raise ValueError(f"Invalid content type: {type(content)}") + @staticmethod + def _transform_responses_api_content_part_to_chat_completion_part( + fields: dict[str, object], + ) -> dict[str, object] | None: + if (media := decode_responses_media(fields)) is not None: + return dict(media) + part_type: Final = fields.get("type") + if part_type == "input_image": + return dict(_responses_image_part(fields)) + if part_type == "input_file": + return dict(_responses_file_part(fields)) + if part_type == "input_audio": + return dict(_validate_responses_media(_AUDIO_PART_ADAPTER, fields)) + if part_type == "refusal": + refusal: Final = fields.get("refusal") + return dict(OpenAIChatCompletionTextObject(type="text", text=refusal)) if isinstance(refusal, str) else None + if part_type == "encrypted_content": + encrypted_content: Final = fields.get("encrypted_content") + return ( + None + if encrypted_content is None + else dict(OpenAIChatCompletionTextObject(type="text", text=str(encrypted_content))) + ) + text_value: Final = fields.get("text") + if text_value is None: + return None + return { + "type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type( + part_type if isinstance(part_type, str) and part_type else "text" + ), + "text": text_value, + **({"cache_control": fields["cache_control"]} if "cache_control" in fields else {}), + } + @staticmethod def _get_chat_completion_request_content_type( content_type: str, @@ -3045,6 +3082,18 @@ class LiteLLMCompletionResponsesConfig: # Default to completed for unknown finish reasons return "completed" + @staticmethod + def _is_paused_turn(choice: Choices) -> bool: + return anthropic_stop_reason(choice.message) == "pause_turn" + + @staticmethod + def _responses_status_for_choice(choice: Choices) -> ResponsesAPIStatus: + if LiteLLMCompletionResponsesConfig._is_paused_turn(choice): + return "completed" + return LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status( + choice.finish_reason + ) + @staticmethod def _incomplete_details_for_finish_reason( finish_reason: str | None, @@ -3181,19 +3230,21 @@ class LiteLLMCompletionResponsesConfig: """ if isinstance(chat_completion_response, dict): chat_completion_response = ModelResponse(**chat_completion_response) - # Get finish_reason from the first choice to determine overall status - finish_reason: str | None = None choices: Final[list[Choices]] = getattr(chat_completion_response, "choices", []) - if choices and len(choices) > 0: - finish_reason = choices[0].finish_reason + finish_reason: Final = choices[0].finish_reason if choices else None + paused_turn: Final = bool(choices) and LiteLLMCompletionResponsesConfig._is_paused_turn(choices[0]) echoed: Final = _echoable_request_params(responses_api_request) chat_usage: Final = getattr(chat_completion_response, "usage", None) - incomplete_details: Final = LiteLLMCompletionResponsesConfig._incomplete_details_for_finish_reason( - finish_reason=finish_reason, - existing=getattr(chat_completion_response, "incomplete_details", None), - completion_tokens=getattr(chat_usage, "completion_tokens", None), - max_output_tokens=echoed.get("max_output_tokens"), + incomplete_details: Final = ( + None + if paused_turn + else LiteLLMCompletionResponsesConfig._incomplete_details_for_finish_reason( + finish_reason=finish_reason, + existing=getattr(chat_completion_response, "incomplete_details", None), + completion_tokens=getattr(chat_usage, "completion_tokens", None), + max_output_tokens=echoed.get("max_output_tokens"), + ) ) responses_api_response: Final[ResponsesAPIResponse] = ResponsesAPIResponse( @@ -3220,8 +3271,8 @@ class LiteLLMCompletionResponsesConfig: max_output_tokens=echoed.get("max_output_tokens"), previous_response_id=echoed.get("previous_response_id"), reasoning=echoed.get("reasoning"), - status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status( - finish_reason + status=( + LiteLLMCompletionResponsesConfig._responses_status_for_choice(choices[0]) if choices else "incomplete" ), text=echoed.get("text") or {}, truncation=echoed.get("truncation"), @@ -3367,9 +3418,7 @@ class LiteLLMCompletionResponsesConfig: GenericResponseOutputItem( type="reasoning", id=mint_reasoning_item_id(), - status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status( - choice.finish_reason - ), + status=LiteLLMCompletionResponsesConfig._responses_status_for_choice(choice), role="assistant", content=[ OutputText( @@ -3492,24 +3541,34 @@ class LiteLLMCompletionResponsesConfig: choice=choice, ) message_output_items.extend(image_generation_items) - elif choice.message.content is not None or getattr(choice.message, "annotations", None): + elif content := LiteLLMCompletionResponsesConfig._message_output_content(choice.message): message_output_items.append( GenericResponseOutputItem( type="message", id=f"msg_{uuid.uuid4()}", - status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status( - choice.finish_reason - ), + status=LiteLLMCompletionResponsesConfig._responses_status_for_choice(choice), role=choice.message.role, - content=[ - LiteLLMCompletionResponsesConfig._transform_chat_message_to_response_output_text( - choice.message - ) - ], + content=list(content), ) ) return message_output_items + @staticmethod + def _message_output_content(message: Message) -> tuple[OutputText | ResponseOutputRefusal, ...]: + refusal: Final = openai_chat_refusal_text(message) + refusal_parts: Final = (ResponseOutputRefusal(type="refusal", refusal=refusal),) if refusal is not None else () + has_text_part: Final = ( + bool(message.content) + or bool(getattr(message, "annotations", None)) + or (message.content is not None and not refusal_parts) + ) + text_parts: Final = ( + (LiteLLMCompletionResponsesConfig._transform_chat_message_to_response_output_text(message),) + if has_text_part + else () + ) + return (*text_parts, *refusal_parts) + @staticmethod def _transform_responses_api_outputs_to_chat_completion_messages( responses_api_output: ResponsesAPIResponse, diff --git a/litellm/types/llms/anthropic.py b/litellm/types/llms/anthropic.py index 152614e4385..dc35af81091 100644 --- a/litellm/types/llms/anthropic.py +++ b/litellm/types/llms/anthropic.py @@ -751,7 +751,9 @@ class AnthropicOutputTokensDetails(LiteLLMBaseModel): thinking_tokens: int | None = None -AnthropicFinishReason = Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction"] +AnthropicFinishReason = Literal[ + "end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "pause_turn", "compaction" +] class AnthropicResponse(LiteLLMBaseModel): diff --git a/litellm/types/llms/anthropic_messages/anthropic_response.py b/litellm/types/llms/anthropic_messages/anthropic_response.py index 9ad13243451..4f419385630 100644 --- a/litellm/types/llms/anthropic_messages/anthropic_response.py +++ b/litellm/types/llms/anthropic_messages/anthropic_response.py @@ -98,7 +98,7 @@ class AnthropicMessagesResponse(TypedDict, total=False): model: str | None # This represents the Model type from Anthropic role: Literal["assistant"] | None stop_reason: ReadOnly[ - Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction"] | None + Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction", "pause_turn"] | None ] stop_details: NotRequired[ReadOnly[AnthropicStopDetails | None]] stop_sequence: str | None diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index ee86ec52327..d7e55b57a07 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -70,6 +70,7 @@ from pydantic import ( FieldSerializationInfo, NonNegativeInt, PrivateAttr, + SerializeAsAny, SerializerFunctionWrapHandler, field_serializer, field_validator, @@ -1855,7 +1856,7 @@ class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject): item_id: str output_index: int content_index: int - part: BaseLiteLLMOpenAIResponseObject + part: SerializeAsAny[BaseLiteLLMOpenAIResponseObject] class ContentPartDonePartOutputText(BaseLiteLLMOpenAIResponseObject): diff --git a/litellm/types/responses/main.py b/litellm/types/responses/main.py index 72dc63a581f..86dc857cfd0 100644 --- a/litellm/types/responses/main.py +++ b/litellm/types/responses/main.py @@ -4,6 +4,7 @@ from typing import Final, Literal, Optional, Union from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall from openai.types.responses.response_function_web_search import ActionSearchSource, ResponseFunctionWebSearch +from openai.types.responses.response_output_refusal import ResponseOutputRefusal from pydantic import ConfigDict, PrivateAttr from typing_extensions import TypedDict @@ -154,7 +155,7 @@ class GenericResponseOutputItem(BaseLiteLLMOpenAIResponseObject): id: str status: str # "completed", "in_progress", etc. role: str # "assistant", "user", etc. - content: list[OutputText] + content: list[OutputText | ResponseOutputRefusal] phase: Phase = None diff --git a/tests/integration/providers/test_anthropic_context_window_stop_wire.py b/tests/integration/providers/test_anthropic_context_window_stop_wire.py index 38e3e7dffcd..3e268b89ce2 100644 --- a/tests/integration/providers/test_anthropic_context_window_stop_wire.py +++ b/tests/integration/providers/test_anthropic_context_window_stop_wire.py @@ -9,6 +9,15 @@ from integration._support.wire import Reply, Request, wire_server from pydantic import JsonValue MODEL: Final = "claude-haiku-4-5" +TEXT: Final = {"type": "text", "text": "alpha bravo"} +EXPLANATION: Final = "This request was declined under the usage policy." +PENDING_SEARCH: Final = { + "type": "server_tool_use", + "id": "srvtoolu_pending", + "name": "web_search", + "input": {"query": "next city"}, +} +TERMINAL_EVENTS: Final = ("response.completed", "response.incomplete", "response.failed") def _sse(event: Mapping[str, JsonValue]) -> bytes: @@ -21,36 +30,64 @@ def _message(identity: str, stop_reason: str | None) -> dict[str, JsonValue]: "type": "message", "role": "assistant", "model": MODEL, - "content": [{"type": "text", "text": "alpha bravo"}], + "content": [TEXT], "stop_reason": stop_reason, "stop_sequence": None, "usage": {"input_tokens": 199878, "output_tokens": 12799}, } -def _stream(identity: str, stop_reason: str) -> tuple[bytes, ...]: +def _block_events(index: int, block: Mapping[str, JsonValue]) -> tuple[bytes, ...]: + if block["type"] == "text": + return ( + _sse({"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}), + _sse( + { + "type": "content_block_delta", + "index": index, + "delta": {"type": "text_delta", "text": block["text"]}, + } + ), + _sse({"type": "content_block_stop", "index": index}), + ) return ( - _sse({"type": "message_start", "message": {**_message(identity, None), "content": []}}), - _sse({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}), - _sse({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "alpha bravo"}}), - _sse({"type": "content_block_stop", "index": 0}), + _sse({"type": "content_block_start", "index": index, "content_block": {**block, "input": {}}}), _sse( { - "type": "message_delta", - "delta": {"stop_reason": stop_reason, "stop_sequence": None}, - "usage": {"output_tokens": 12799}, + "type": "content_block_delta", + "index": index, + "delta": {"type": "input_json_delta", "partial_json": json.dumps(block["input"])}, } ), - _sse({"type": "message_stop"}), + _sse({"type": "content_block_stop", "index": index}), ) -def _anthropic(identity: str, stop_reason: str): +def _turn( + identity: str, + stop_reason: str, + content: tuple[Mapping[str, JsonValue], ...] = (TEXT,), + stop_details: Mapping[str, JsonValue] | None = None, +): + stop: Final = {"stop_reason": stop_reason, "stop_sequence": None} + details: Final = {"stop_details": stop_details} if stop_details is not None else {} + message: Final = {**_message(identity, stop_reason), "content": list(content), **details} + stream: Final = ( + _sse({"type": "message_start", "message": {**_message(identity, None), "content": []}}), + *( + event + for index, block in enumerate(content) + for event in _block_events(index, block) # comprehension-ok: block -> events flatten + ), + _sse({"type": "message_delta", "delta": {**stop, **details}, "usage": {"output_tokens": 12799}}), + _sse({"type": "message_stop"}), + ) + def respond(request: Request) -> Reply: assert request.method == "POST" and request.target == "/v1/messages", request.target if json.loads(request.body).get("stream"): - return Reply(content_type="text/event-stream", chunks=_stream(identity, stop_reason)) - return Reply(body=json.dumps(_message(identity, stop_reason)).encode()) + return Reply(content_type="text/event-stream", chunks=stream) + return Reply(body=json.dumps(message).encode()) return respond @@ -67,45 +104,70 @@ def _events(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple ) -def _chat_finish_reason(gateway: Gateway, model: str, *, stream: bool) -> JsonValue: - body: Final = {"model": model, "max_tokens": 64000, "messages": [{"role": "user", "content": "copy the list"}]} +def _chat_choices(gateway: Gateway, body: Mapping[str, JsonValue], *, stream: bool) -> tuple[dict[str, JsonValue], ...]: if stream: - reasons: Final = tuple( - choice["finish_reason"] + return tuple( + choice for chunk in _events(gateway, "/v1/chat/completions", body) for choice in chunk["choices"] # comprehension-ok: chunk -> choices flatten - if choice.get("finish_reason") ) - assert len(reasons) == 1, reasons - return reasons[0] response: Final = gateway.request("POST", "/v1/chat/completions", body) assert response.status_code == 200, response.text - return response.json()["choices"][0]["finish_reason"] + return tuple(response.json()["choices"]) -def _responses_status(gateway: Gateway, model: str, *, stream: bool) -> JsonValue: - body: Final = {"model": model, "max_output_tokens": 64000, "input": "copy the list"} +def _chat_body(model: str) -> dict[str, JsonValue]: + return {"model": model, "max_tokens": 64000, "messages": [{"role": "user", "content": "copy the list"}]} + + +def _chat_finish_reason(gateway: Gateway, model: str, *, stream: bool) -> JsonValue: + reasons: Final = tuple( + choice["finish_reason"] + for choice in _chat_choices(gateway, _chat_body(model), stream=stream) + if choice.get("finish_reason") + ) + assert len(reasons) == 1, reasons + return reasons[0] + + +def _chat_messages(gateway: Gateway, model: str, *, stream: bool) -> tuple[dict[str, JsonValue], ...]: + return tuple( + choice.get("delta") or choice.get("message") or {} + for choice in _chat_choices(gateway, _chat_body(model), stream=stream) + ) + + +def _responses_body(model: str) -> dict[str, JsonValue]: + return {"model": model, "max_output_tokens": 64000, "input": "copy the list", "tools": [{"type": "web_search"}]} + + +def _responses_result(gateway: Gateway, model: str, *, stream: bool) -> dict[str, JsonValue]: if stream: terminal: Final = tuple( event["response"] - for event in _events(gateway, "/v1/responses", body) - if event["type"] in ("response.completed", "response.incomplete", "response.failed") + for event in _events(gateway, "/v1/responses", _responses_body(model)) + if event["type"] in TERMINAL_EVENTS ) assert len(terminal) == 1, terminal - return terminal[0]["status"] - response: Final = gateway.request("POST", "/v1/responses", body) + return terminal[0] + response: Final = gateway.request("POST", "/v1/responses", _responses_body(model)) assert response.status_code == 200, response.text - return response.json()["status"] + return response.json() @pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) @pytest.mark.parametrize( ("stop_reason", "finish_reason"), - [("model_context_window_exceeded", "length"), ("max_tokens", "length"), ("end_turn", "stop")], + [ + ("model_context_window_exceeded", "length"), + ("max_tokens", "length"), + ("pause_turn", "length"), + ("end_turn", "stop"), + ], ) def test_chat_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, finish_reason: str, stream: bool) -> None: identity: Final = uuid.uuid4().hex - with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario: + with wire_server(_turn(identity, stop_reason)) as wire, gateway.scenario() as scenario: model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") assert _chat_finish_reason(gateway, model, stream=stream) == finish_reason @@ -113,10 +175,81 @@ def test_chat_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, fin @pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) @pytest.mark.parametrize( ("stop_reason", "status"), - [("model_context_window_exceeded", "incomplete"), ("max_tokens", "incomplete"), ("end_turn", "completed")], + [ + ("model_context_window_exceeded", "incomplete"), + ("max_tokens", "incomplete"), + ("pause_turn", "completed"), + ("end_turn", "completed"), + ], ) def test_responses_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, status: str, stream: bool) -> None: identity: Final = uuid.uuid4().hex - with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario: + with wire_server(_turn(identity, stop_reason)) as wire, gateway.scenario() as scenario: model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") - assert _responses_status(gateway, model, stream=stream) == status + assert _responses_result(gateway, model, stream=stream)["status"] == status + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_keeps_pause_turn_so_the_client_knows_to_resend_the_turn(gateway: Gateway, stream: bool) -> None: + identity: Final = uuid.uuid4().hex + paused: Final = _turn(identity, "pause_turn", (PENDING_SEARCH,)) + with wire_server(paused) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") + stop_reasons: Final = tuple( + (message.get("provider_specific_fields") or {}).get("stop_reason") + for message in _chat_messages(gateway, model, stream=stream) + ) + assert "pause_turn" in stop_reasons, stop_reasons + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_responses_list_a_paused_turn_unrun_search_as_an_in_progress_web_search(gateway: Gateway, stream: bool) -> None: + identity: Final = uuid.uuid4().hex + paused: Final = _turn(identity, "pause_turn", (PENDING_SEARCH,)) + with wire_server(paused) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") + result: Final = _responses_result(gateway, model, stream=stream) + assert (result["status"], result.get("incomplete_details")) == ("completed", None) + assert [(item["type"], item["status"], item["action"]["query"]) for item in result["output"]] == [ + ("web_search_call", "in_progress", PENDING_SEARCH["input"]["query"]) + ] + + +_REFUSAL: Final = {"type": "refusal", "category": "cyber", "explanation": EXPLANATION} + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +def test_chat_returns_the_refusal_explanation_as_the_message_refusal(gateway: Gateway, stream: bool) -> None: + identity: Final = uuid.uuid4().hex + with wire_server(_turn(identity, "refusal", (), _REFUSAL)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") + refusal: Final = "".join( + message["refusal"] for message in _chat_messages(gateway, model, stream=stream) if message.get("refusal") + ) + assert refusal == EXPLANATION + + +@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"]) +@pytest.mark.parametrize( + ("content", "expected_parts"), + [ + ((), [("refusal", EXPLANATION)]), + ((TEXT,), [("output_text", TEXT["text"]), ("refusal", EXPLANATION)]), + ], + ids=["refusal_only", "answer_then_refusal"], +) +def test_responses_return_the_refusal_explanation_as_a_refusal_part( + gateway: Gateway, + content: tuple[Mapping[str, JsonValue], ...], + expected_parts: list[tuple[str, str]], + stream: bool, +) -> None: + identity: Final = uuid.uuid4().hex + with wire_server(_turn(identity, "refusal", content, _REFUSAL)) as wire, gateway.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key") + result: Final = _responses_result(gateway, model, stream=stream) + assert (result["status"], result["incomplete_details"]) == ("incomplete", {"reason": "content_filter"}) + assert [ + [(part["type"], part.get("text") or part.get("refusal")) for part in item["content"]] + for item in result["output"] + ] == [expected_parts] diff --git a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py index 778c9d68be5..8a755d47817 100644 --- a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py +++ b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py @@ -5383,8 +5383,9 @@ def test_convert_gemini_tool_call_result_answers_tool_reference_only_result(): def test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use(): """A replayed srvtoolu_ call whose server tool result is not available (e.g. the Responses bridge replays items without provider_specific_fields) - must become a plain client tool_use so the client's tool_result can pair - with it. A dangling server_tool_use makes Anthropic 400 the request with + and that the client answered with its own tool message must become a plain + client tool_use so the client's tool_result can pair with it. A dangling + server_tool_use makes Anthropic 400 the request with "unexpected `tool_use_id` found in `tool_result` blocks".""" from litellm.litellm_core_utils.prompt_templates.factory import convert_to_anthropic_tool_invoke @@ -5398,6 +5399,7 @@ def test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use(): ], web_search_results=None, tool_results=None, + answered_tool_ids=frozenset({"srvtoolu_01Unpaired"}), ) assert result == [ @@ -7435,17 +7437,11 @@ def test_convert_to_anthropic_tool_invoke_sanitizes_invalid_ids(): def test_convert_to_anthropic_tool_invoke_server_tool(): """ - Test that a server tool call (srvtoolu_) with no stored result is replayed - as a regular tool_use block. - - A server_tool_use block is only valid when paired with its result block, so - an unpaired one must degrade to tool_use for Anthropic to accept the replay. - A paired call still becomes server_tool_use, covered by - test_convert_to_anthropic_tool_invoke_with_web_search_results. - - Context: https://github.com/BerriAI/litellm/issues/17737 (original - server_tool_use reconstruction) and LIT-6622 / PR #39144 (unpaired calls - degrade instead of 400ing at Anthropic). + A server tool call (srvtoolu_) with no stored result and no client answer is + the unrun call a pause_turn response ends on, so it is replayed as a + server_tool_use for Anthropic to run when the client resends the turn. + An answered one degrades to tool_use, covered by + test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use. """ tool_calls = [ { @@ -7460,11 +7456,14 @@ def test_convert_to_anthropic_tool_invoke_server_tool(): result = convert_to_anthropic_tool_invoke(tool_calls) - assert len(result) == 1 - assert result[0]["type"] == "tool_use" - assert result[0]["id"] == "srvtoolu_01ABC123" - assert result[0]["name"] == "web_search" - assert result[0]["input"] == {"query": "elephant weight"} + assert result == [ + { + "type": "server_tool_use", + "id": "srvtoolu_01ABC123", + "name": "web_search", + "input": {"query": "elephant weight"}, + } + ] def test_convert_to_anthropic_tool_invoke_with_web_search_results(): @@ -8231,6 +8230,170 @@ def test_anthropic_messages_pt_interleave_more_thinking_than_tool_groups(): ], f"Expected order but got: {types}" + +def _paused_web_search_turn( + thinking_blocks: list[dict[str, str]] | None, content: str | list[dict[str, str]] = "Step 1 is Tokyo." +) -> dict[str, object]: + assistant_message: Final[dict[str, object]] = { + "role": "assistant", + "content": content, + "tool_calls": [ + { + "id": "srvtoolu_01Done", + "type": "function", + "function": {"name": "web_search", "arguments": '{"query": "Tokyo sister cities"}'}, + }, + { + "id": "srvtoolu_02Done", + "type": "function", + "function": {"name": "web_search", "arguments": '{"query": "Adana sister cities"}'}, + }, + { + "id": "srvtoolu_01Paused", + "type": "function", + "function": {"name": "web_search", "arguments": '{"query": "Adelaide sister cities"}'}, + }, + ], + "provider_specific_fields": { + "stop_reason": "pause_turn", + "web_search_results": [ + { + "type": "web_search_tool_result", + "tool_use_id": tool_use_id, + "content": [ + {"type": "web_search_result", "url": "https://example.com", "title": "City", "snippet": "x"} + ], + } + for tool_use_id in ("srvtoolu_01Done", "srvtoolu_02Done") + ], + }, + } + if thinking_blocks is None: + return assistant_message + return {**assistant_message, "thinking_blocks": thinking_blocks} + + +_PAUSED_TURN_THINKING_CASES = pytest.mark.parametrize( + "thinking_blocks", + [ + None, + [{"type": "thinking", "thinking": "Search first", "signature": "sig_1"}], + [ + {"type": "thinking", "thinking": "Search first", "signature": "sig_1"}, + {"type": "thinking", "thinking": "Search again", "signature": "sig_2"}, + {"type": "thinking", "thinking": "Search once more", "signature": "sig_3"}, + {"type": "thinking", "thinking": "Wrap up later", "signature": "sig_4"}, + ], + ], + ids=["without_thinking", "fewer_thinking_than_searches", "more_thinking_than_searches"], +) + +_PAUSED_TURN_CONTENT_CASES = pytest.mark.parametrize( + "content", + [ + "Step 1 is Tokyo.", + [{"type": "text", "text": "Step 1 is Tokyo."}], + [ + {"type": "thinking", "thinking": "Search first", "signature": "sig_1"}, + {"type": "text", "text": "Step 1 is Tokyo."}, + ], + ], + ids=["string_content", "text_parts", "text_parts_with_inline_thinking"], +) + + +@_PAUSED_TURN_THINKING_CASES +@_PAUSED_TURN_CONTENT_CASES +def test_anthropic_messages_pt_paused_turn_ends_on_unrun_server_tool_use(thinking_blocks, content): + """A pause_turn response resent as-is must end on its unrun server_tool_use: + Anthropic reads a turn ending on text as prefill, which newer models reject, + and a tool_use without a tool_result is a 400. Its thinking blocks are dropped, + since chat cannot say where Claude wrote them and Anthropic rejects moved ones.""" + result = anthropic_messages_pt( + [ + {"role": "user", "content": "Play the sister city chain game."}, + _paused_web_search_turn(thinking_blocks, content), + ], + model="claude-sonnet-4-5", + llm_provider="anthropic", + ) + + replayed = result[-1]["content"] + assert replayed[-1] == { + "type": "server_tool_use", + "id": "srvtoolu_01Paused", + "name": "web_search", + "input": {"query": "Adelaide sister cities"}, + } + assert [block["id"] for block in replayed if block["type"] == "server_tool_use"] == [ + "srvtoolu_01Done", + "srvtoolu_02Done", + "srvtoolu_01Paused", + ] + assert [block["type"] for block in replayed if block["type"] in ("tool_use", "thinking")] == [] + assert [block.get("tool_use_id") for block in replayed if block["type"] == "web_search_tool_result"] == [ + "srvtoolu_01Done", + "srvtoolu_02Done", + ] + + +@_PAUSED_TURN_THINKING_CASES +@_PAUSED_TURN_CONTENT_CASES +def test_anthropic_messages_pt_paused_turn_before_new_user_message_drops_unrun_search(thinking_blocks, content): + """A client may answer a paused turn with a new user message instead of resending it. + Anthropic then rejects a server_tool_use without its result, so the unrun search is dropped.""" + result = anthropic_messages_pt( + [ + {"role": "user", "content": "Play the sister city chain game."}, + _paused_web_search_turn(thinking_blocks, content), + {"role": "user", "content": "Please continue."}, + ], + model="claude-sonnet-4-5", + llm_provider="anthropic", + ) + + assert [message["role"] for message in result] == ["user", "assistant", "user"] + replayed = result[1]["content"] + assert [block["id"] for block in replayed if block["type"] == "server_tool_use"] == [ + "srvtoolu_01Done", + "srvtoolu_02Done", + ] + assert [block["type"] for block in replayed if block["type"] in ("tool_use", "thinking")] == [] + assert [block.get("tool_use_id") for block in replayed if block["type"] == "web_search_tool_result"] == [ + "srvtoolu_01Done", + "srvtoolu_02Done", + ] + assert {"type": "text", "text": "Step 1 is Tokyo."} in replayed + + +@_PAUSED_TURN_THINKING_CASES +@_PAUSED_TURN_CONTENT_CASES +def test_anthropic_messages_pt_paused_turn_resent_without_search_results_runs_only_its_last_search( + thinking_blocks, content +): + paused_turn = _paused_web_search_turn(thinking_blocks, content) + result = anthropic_messages_pt( + [ + {"role": "user", "content": "Play the sister city chain game."}, + {key: value for key, value in paused_turn.items() if key != "provider_specific_fields"}, + ], + model="claude-sonnet-4-5", + llm_provider="anthropic", + ) + + replayed = result[-1]["content"] + assert [block["id"] for block in replayed if block["type"] in ("server_tool_use", "tool_use")] == [ + "srvtoolu_01Paused" + ] + assert replayed[-1] == { + "type": "server_tool_use", + "id": "srvtoolu_01Paused", + "name": "web_search", + "input": {"query": "Adelaide sister cities"}, + } + assert [block["type"] for block in replayed if block["type"] == "thinking"] == [] + + def test_anthropic_messages_pt_list_content_with_thinking_preserves_order(): """ Test that when assistant content is already a list containing interleaved diff --git a/tests/unit/litellm_core_utils/test_core_helpers.py b/tests/unit/litellm_core_utils/test_core_helpers.py index 93defc0f04f..1a92aa7c2c1 100644 --- a/tests/unit/litellm_core_utils/test_core_helpers.py +++ b/tests/unit/litellm_core_utils/test_core_helpers.py @@ -265,6 +265,7 @@ class TestMapFinishReasonAnthropic: ("max_tokens", "length"), ("tool_use", "tool_calls"), ("compaction", "length"), + ("pause_turn", "length"), ("model_context_window_exceeded", "length"), ("content_filtered", "content_filter"), ], diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py index 5d3fc523c0c..95068008d25 100644 --- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py +++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py @@ -3651,6 +3651,51 @@ def test_streamed_refusal_stop_details_reach_the_assembled_message(): assert provider_fields["stop_details"] == stop_details +def test_streamed_refusal_explanation_reaches_the_client_as_the_message_refusal(): + stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"] + assembled: Final = _assemble_stream(_message_stream_events("refusal", stop_details)) + + assert assembled.choices[0].message.refusal == stop_details["explanation"] + + +def test_streamed_refusal_without_an_explanation_sets_no_message_refusal(): + assembled: Final = _assemble_stream( + _message_stream_events("refusal", {"type": "refusal", "category": "cyber", "explanation": None}) + ) + + assert assembled.choices[0].message.get("refusal") is None + + +def _paused_search_stream_events() -> tuple[dict, ...]: + message_start, message_delta, message_stop = _message_stream_events("pause_turn", None) + return ( + message_start, + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "server_tool_use", "id": "srvtoolu_pending", "name": "web_search", "input": {}}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "input_json_delta", "partial_json": '{"query": "next city"}'}, + }, + {"type": "content_block_stop", "index": 0}, + {**message_delta, "delta": {"stop_reason": "pause_turn", "stop_sequence": None}}, + message_stop, + ) + + +def test_streamed_pause_turn_is_an_unfinished_turn_listing_the_query_it_will_run(): + assembled: Final = _assemble_stream(_paused_search_stream_events()) + + assert assembled.choices[0].finish_reason == "length" + provider_fields: Final = assembled.choices[0].message.provider_specific_fields + assert provider_fields["stop_reason"] == "pause_turn" + pending: Final = provider_fields["web_search_calls"]["srvtoolu_pending"] + assert (pending.status, pending.action.query) == ("in_progress", "next city") + + def test_streamed_end_turn_carries_no_stop_details(): assembled: Final = _assemble_stream(_message_stream_events("end_turn", None)) diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py index 501dd5bd0c6..c8a2f783809 100644 --- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py +++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py @@ -9715,6 +9715,75 @@ def test_end_turn_response_carries_no_stop_details(): assert "stop_reason" not in provider_fields +def test_refusal_explanation_reaches_the_client_as_the_message_refusal(): + result: Final = _transform_message_body(ANTHROPIC_DOCS_REFUSAL_MESSAGE) + + assert result.choices[0].message.refusal == ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]["explanation"] + + +@pytest.mark.parametrize( + "stop_details", + [None, {"type": "refusal", "category": "cyber", "explanation": None}, {"type": "refusal", "explanation": ""}], + ids=["no_stop_details", "null_explanation", "empty_explanation"], +) +def test_refusal_without_an_explanation_sets_no_message_refusal(stop_details: dict | None): + result: Final = _transform_message_body(_refusal_message_body("refusal", stop_details)) + + assert result.choices[0].message.get("refusal") is None + + +_RAN_SEARCH: Final = { + "type": "server_tool_use", + "id": "srvtoolu_ran", + "name": "web_search", + "input": {"query": "tokyo"}, +} +_RAN_SEARCH_RESULT: Final = { + "type": "web_search_tool_result", + "tool_use_id": "srvtoolu_ran", + "content": [ + {"type": "web_search_result", "url": "https://tokyo.example", "title": "Tokyo", "encrypted_content": "e"} + ], +} +_PENDING_SEARCH: Final = { + "type": "server_tool_use", + "id": "srvtoolu_pending", + "name": "web_search", + "input": {"query": "next city"}, +} + + +def _paused_turn(*content: dict) -> dict: + return {**_refusal_message_body("pause_turn", None), "content": list(content)} + + +def test_pause_turn_is_an_unfinished_turn_that_keeps_its_native_stop_reason(): + result: Final = _transform_message_body(_paused_turn(_RAN_SEARCH, _RAN_SEARCH_RESULT, _PENDING_SEARCH)) + + assert result.choices[0].finish_reason == "length" + assert result.choices[0].message.provider_specific_fields["stop_reason"] == "pause_turn" + + +@pytest.mark.parametrize( + ("content", "expected_calls"), + [ + ( + (_RAN_SEARCH, _RAN_SEARCH_RESULT, _PENDING_SEARCH), + [("ws_srvtoolu_ran", "completed", "tokyo"), ("ws_srvtoolu_pending", "in_progress", "next city")], + ), + ((_PENDING_SEARCH,), [("ws_srvtoolu_pending", "in_progress", "next city")]), + ], + ids=["after_a_finished_search", "before_any_search_result"], +) +def test_pause_turn_lists_the_search_it_has_not_run_yet_as_in_progress( + content: tuple[dict, ...], expected_calls: list[tuple[str, str, str]] +): + result: Final = _transform_message_body(_paused_turn(*content)) + + calls: Final = result.choices[0].message.provider_specific_fields["web_search_calls"] + assert [(call.id, call.status, call.action.query) for call in calls] == expected_calls + + def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict: return { **_refusal_message_body(stop_reason, stop_details), @@ -9741,6 +9810,7 @@ def test_json_mode_refusal_keeps_stop_details_on_the_message(): message: Final = result.choices[0].message assert json.loads(message.content) == {"answer": "Hello"} assert message.provider_specific_fields == {"stop_reason": "refusal", "stop_details": stop_details} + assert message.refusal == stop_details["explanation"] def test_json_mode_tool_use_turn_adds_no_stop_fields_to_the_message(): diff --git a/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py b/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py index e0882653324..cc366dcde4c 100644 --- a/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py +++ b/tests/unit/llms/anthropic/messages/adapters/test_anthropic_messages_adapters_transformation.py @@ -126,6 +126,57 @@ def test_translate_chat_length_takes_precedence_over_refusal(): assert result.get("stop_details") is None +@pytest.mark.parametrize( + ("provider_specific_fields", "expected_stop_reason"), + [({"stop_reason": "pause_turn"}, "pause_turn"), (None, "max_tokens")], + ids=["paused_claude_turn", "truncated_turn"], +) +def test_translate_chat_length_keeps_a_paused_claude_turn_unfinished( + provider_specific_fields: dict[str, object] | None, expected_stop_reason: str +): + """A paused Claude turn reaches chat as ``length``, which a /v1/messages client would read as a cut-off reply.""" + message: Final = Message(content="Searching", role="assistant", provider_specific_fields=provider_specific_fields) + response: Final = ModelResponse(choices=[Choices(index=0, finish_reason="length", message=message)], usage=Usage()) + + result: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response) + + assert result["stop_reason"] == expected_stop_reason + assert result["content"] == [{"type": "text", "text": "Searching"}] + + +@pytest.mark.parametrize( + ("stop_details", "expected_stop_details"), + [ + ( + {"type": "refusal", "category": "cyber", "explanation": "I can't help with that."}, + {"type": "refusal", "category": "cyber", "explanation": "I can't help with that."}, + ), + ( + {"type": "refusal", "category": 7, "explanation": "I can't help with that."}, + {"type": "refusal", "category": None, "explanation": "I can't help with that."}, + ), + ], + ids=["claude_stop_details", "unreadable_stop_details"], +) +def test_translate_claude_refusal_keeps_the_explanation_out_of_the_reply_text(stop_details, expected_stop_details): + """Claude's refusal explanation is API metadata, so replaying it as a text block puts words in Claude's mouth.""" + message: Final = Message( + content="Partial answer", + role="assistant", + refusal=stop_details["explanation"], + provider_specific_fields={"stop_reason": "refusal", "stop_details": stop_details}, + ) + response: Final = ModelResponse( + choices=[Choices(index=0, finish_reason="content_filter", message=message)], usage=Usage() + ) + + result: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response) + + assert result["content"] == [{"type": "text", "text": "Partial answer"}] + assert result["stop_reason"] == "refusal" + assert result.get("stop_details") == expected_stop_details + + def test_translate_chat_content_filter_to_anthropic_response(): response = ModelResponse( id="chatcmpl-content-filter", diff --git a/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py b/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py index a3cb6ed98f4..79cde87b4e0 100644 --- a/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py +++ b/tests/unit/llms/anthropic/messages/adapters/test_streaming_iterator_first_delta.py @@ -241,6 +241,57 @@ async def test_streaming_chat_length_takes_precedence_over_refusal(async_mode: b assert "stop_details" not in message_delta["delta"] +CLAUDE_REFUSAL_STOP_DETAILS: Final = MappingProxyType( + {"type": "refusal", "category": "cyber", "explanation": "I can't help with that."} +) + + +def _claude_chat_stream( + stop_fields: Mapping[str, object], finish_reason: str, refusal: str | None = None +) -> list[MagicMock]: + """The chunks the chat stream hands the adapter for a Claude turn: the stop fields + ride an empty delta ahead of the chunk that carries the finish_reason.""" + return [ + _make_chunk(Delta(content="Partial answer")), + _make_chunk(Delta(content="", refusal=refusal, provider_specific_fields=dict(stop_fields))), + _make_chunk(Delta(content=None), finish_reason=finish_reason), + ] + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_streaming_claude_refusal_keeps_the_explanation_in_stop_details_only(async_mode: bool): + """Claude's refusal explanation is API metadata, so streaming it as text puts words in Claude's mouth.""" + chunks: Final = _claude_chat_stream( + {"stop_reason": "refusal", "stop_details": dict(CLAUDE_REFUSAL_STOP_DETAILS)}, + "content_filter", + refusal=CLAUDE_REFUSAL_STOP_DETAILS["explanation"], + ) + stream: Final = _AsyncStream(chunks) if async_mode else iter(chunks) + wrapper: Final = AnthropicStreamWrapper(completion_stream=stream, model="claude-model") + + events: Final = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper) + + assert _text_deltas(events) == ["Partial answer"] + message_delta: Final = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"] == {"stop_reason": "refusal", "stop_details": dict(CLAUDE_REFUSAL_STOP_DETAILS)} + + +@pytest.mark.parametrize("async_mode", [False, True]) +@pytest.mark.asyncio +async def test_streaming_paused_claude_turn_ends_on_pause_turn(async_mode: bool): + """A paused Claude turn reaches chat as ``length``, which a /v1/messages client would read as a cut-off reply.""" + chunks: Final = _claude_chat_stream({"stop_reason": "pause_turn"}, "length") + stream: Final = _AsyncStream(chunks) if async_mode else iter(chunks) + wrapper: Final = AnthropicStreamWrapper(completion_stream=stream, model="claude-model") + + events: Final = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper) + + assert _text_deltas(events) == ["Partial answer"] + message_delta: Final = next(event for event in events if event["type"] == "message_delta") + assert message_delta["delta"] == {"stop_reason": "pause_turn"} + + def _input_json_deltas(events: List[dict]) -> List[str]: return [ e["delta"]["partial_json"] diff --git a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py index d732d922375..3eb5f544ac6 100644 --- a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py +++ b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py @@ -25,6 +25,9 @@ from openai.types.responses import ( ResponseCustomToolCallInputDeltaEvent, ResponseCustomToolCallInputDoneEvent, ResponseFunctionToolCall, + ResponseOutputMessage, + ResponseOutputRefusal, + ResponseOutputText, ) from litellm.integrations.custom_guardrail import CustomGuardrail @@ -39,7 +42,16 @@ from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import BedrockGuardrail from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail from litellm.types.guardrails import ContentFilterAction, ContentFilterPattern -from litellm.types.llms.openai import ChatCompletionToolCallChunk +from litellm.types.llms.openai import ( + ChatCompletionToolCallChunk, + ContentPartDoneEvent, + ContentPartDonePartRefusal, + OutputItemDoneEvent, + RefusalDeltaEvent, + RefusalDoneEvent, + ResponseIncompleteEvent, + ResponsesAPIStreamEvents, +) from litellm.responses.litellm_completion_transformation.compaction_items import compaction_summary from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, @@ -5017,3 +5029,183 @@ class TestCompactionOutputSummaryReachesOutputGuardrails: ) assert ended_key(self.SUMMARY) != ended_key("The user lives elsewhere.") + + +_REFUSAL: Final = "I can't share where Jane lives." +_MASKED_REFUSAL: Final = "I can't share where [NAME] lives." + + +class NameMaskingGuardrail(CustomGuardrail): + def __init__(self, guardrail_name: str) -> None: + super().__init__(guardrail_name=guardrail_name) + self.seen_texts: list[list[str]] = [] + + async def apply_guardrail( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict[str, object], + input_type: Literal["request", "response"], + logging_obj: LiteLLMLoggingObj | None = None, + ) -> GenericGuardrailAPIInputs: + texts: Final = list(inputs.get("texts", [])) + self.seen_texts.append(texts) + return {**inputs, "texts": [text.replace("Jane", "[NAME]") for text in texts]} + + +def _refused_message(*leading_text: str) -> GenericResponseOutputItem: + return GenericResponseOutputItem( + type="message", + id="msg_123", + status="incomplete", + role="assistant", + content=[ + *(OutputText(type="output_text", text=text, annotations=[]) for text in leading_text), + ResponseOutputRefusal(type="refusal", refusal=_REFUSAL), + ], + ) + + +def _refused_response(*leading_text: str) -> ResponsesAPIResponse: + return ResponsesAPIResponse( + id="resp_123", + created_at=1234567890, + model="claude-opus-5-5", + object="response", + status="incomplete", + output=[_refused_message(*leading_text)], + ) + + +class TestRefusalPartsReachTextGuardrails: + @pytest.mark.asyncio + async def test_input_refusal_part_is_scanned_and_rewritten(self) -> None: + guardrail: Final = NameMaskingGuardrail(guardrail_name="test") + data: Final = { + "model": "claude-opus-5-5", + "input": [ + {"role": "user", "content": "Where does Jane live?"}, + {"type": "message", "role": "assistant", "content": [{"type": "refusal", "refusal": _REFUSAL}]}, + ], + } + + result: Final = await OpenAIResponsesHandler().process_input_messages(data, guardrail) + + assert guardrail.seen_texts == [["Where does Jane live?", _REFUSAL]] + assert result["input"][1]["content"] == [{"type": "refusal", "refusal": _MASKED_REFUSAL}] + + @pytest.mark.asyncio + @pytest.mark.parametrize( + ("leading_text", "expected_parts"), + [ + ((), [("refusal", _MASKED_REFUSAL)]), + (("Jane asked first.",), [("output_text", "[NAME] asked first."), ("refusal", _MASKED_REFUSAL)]), + ], + ids=["refusal_only", "text_then_refusal"], + ) + async def test_output_refusal_part_is_scanned_and_rewritten( + self, leading_text: tuple[str, ...], expected_parts: list[tuple[str, str]] + ) -> None: + guardrail: Final = NameMaskingGuardrail(guardrail_name="test") + + result: Final = await OpenAIResponsesHandler().process_output_response( + _refused_response(*leading_text), guardrail + ) + + assert guardrail.seen_texts == [[*leading_text, _REFUSAL]] + assert [ + (part["type"], part.get("text") or part.get("refusal")) for part in result.model_dump()["output"][0]["content"] + ] == expected_parts + + @pytest.mark.asyncio + async def test_output_refusal_part_in_an_openai_message_is_rewritten(self) -> None: + guardrail: Final = NameMaskingGuardrail(guardrail_name="test") + openai_message: Final = ResponseOutputMessage( + type="message", + id="msg_123", + role="assistant", + status="incomplete", + content=[ + ResponseOutputText(type="output_text", text="Jane asked first.", annotations=[]), + ResponseOutputRefusal(type="refusal", refusal=_REFUSAL), + ], + ) + + result: Final = await OpenAIResponsesHandler().process_output_response( + _refused_response().model_copy(update={"output": [openai_message]}), guardrail + ) + + assert guardrail.seen_texts == [["Jane asked first.", _REFUSAL]] + assert [ + (part["type"], part.get("text") or part.get("refusal")) for part in result.model_dump()["output"][0]["content"] + ] == [("output_text", "[NAME] asked first."), ("refusal", _MASKED_REFUSAL)] + + @pytest.mark.asyncio + async def test_streamed_refusal_events_carry_the_rewrite(self) -> None: + position: Final = {"item_id": "msg_123", "output_index": 0, "content_index": 0} + events: Final = [ + RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_REFUSAL, **position), + RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_REFUSAL, **position), + ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + part=ContentPartDonePartRefusal(type="refusal", refusal=_REFUSAL), + **position, + ), + OutputItemDoneEvent(type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, item=_refused_message()), + ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=_refused_response()), + ] + + await OpenAIResponsesHandler().process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=NameMaskingGuardrail(guardrail_name="test"), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert [json.loads(event.model_dump_json()) for event in events] == [ + json.loads(event.model_dump_json()) + for event in ( + RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_MASKED_REFUSAL, **position), + RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_MASKED_REFUSAL, **position), + ContentPartDoneEvent( + type=ResponsesAPIStreamEvents.CONTENT_PART_DONE, + part=ContentPartDonePartRefusal(type="refusal", refusal=_MASKED_REFUSAL), + **position, + ), + OutputItemDoneEvent( + type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + output_index=0, + item=_refused_message().model_copy( + update={"content": [ResponseOutputRefusal(type="refusal", refusal=_MASKED_REFUSAL)]} + ), + ), + ResponseIncompleteEvent( + type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, + response=_refused_response().model_copy( + update={ + "output": [ + _refused_message().model_copy( + update={"content": [ResponseOutputRefusal(type="refusal", refusal=_MASKED_REFUSAL)]} + ) + ] + } + ), + ), + ) + ] + + @pytest.mark.asyncio + async def test_cut_refusal_stream_carries_the_rewrite(self) -> None: + position: Final = {"item_id": "msg_123", "output_index": 0, "content_index": 0} + events: Final = [ + RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_REFUSAL, **position), + RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_REFUSAL, **position), + ] + + await OpenAIResponsesHandler().process_output_streaming_response( + responses_so_far=events, + guardrail_to_apply=NameMaskingGuardrail(guardrail_name="test"), + litellm_logging_obj=None, + deliver_ended_stream_rewrites=True, + ) + + assert (events[0].delta, events[1].refusal) == (_MASKED_REFUSAL, _MASKED_REFUSAL) diff --git a/tests/unit/proxy/guardrails/test_anthropic_sse.py b/tests/unit/proxy/guardrails/test_anthropic_sse.py index 95f596aab11..ff75c8b0110 100644 --- a/tests/unit/proxy/guardrails/test_anthropic_sse.py +++ b/tests/unit/proxy/guardrails/test_anthropic_sse.py @@ -120,8 +120,9 @@ REFUSAL_STOP_DETAILS: Final = {"type": "refusal", "category": "cyber", "explanat [ ("end_turn", {"safeguard_results": SAFEGUARD_RESULTS}), ("refusal", {"stop_details": REFUSAL_STOP_DETAILS}), + ("pause_turn", {}), ], - ids=["safeguard_results", "refusal_stop_details"], + ids=["safeguard_results", "refusal_stop_details", "pause_turn"], ) def test_rewritten_stream_keeps_the_upstream_message_delta_fields(stop_reason: str, delta_fields: dict[str, object]): """A guardrail that rewrites a streamed response assembles it into the OpenAI shape, which has no diff --git a/tests/unit/proxy/policy_engine/test_pipeline_executor.py b/tests/unit/proxy/policy_engine/test_pipeline_executor.py index 27e8dfb7a2c..5a082521660 100644 --- a/tests/unit/proxy/policy_engine/test_pipeline_executor.py +++ b/tests/unit/proxy/policy_engine/test_pipeline_executor.py @@ -1817,6 +1817,94 @@ async def test_streaming_step_passes_a_tool_only_stream_the_legacy_hook_left_alo assert not any("discarded" in record.getMessage() for record in caplog.records) +class _EmailMaskingLegacyHook(CustomGuardrail): + def __init__(self): + super().__init__(guardrail_name="masker", event_hook="post_call", default_on=True) + + async def async_post_call_success_hook(self, data, user_api_key_dict, response): + return { + **response, + "content": [ + {**block, "text": block["text"].replace("bob@example.com", "[EMAIL]")} + if block["type"] == "text" + else block + for block in response["content"] + ], + } + + +def _sse_frame(event, data): + return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode() + + +def _claude_refusal_sse_stream(text): + return [ + _sse_frame( + "message_start", + { + "type": "message_start", + "message": { + "id": "msg_refused", + "type": "message", + "role": "assistant", + "model": "claude-model", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 3, "output_tokens": 1}, + }, + }, + ), + _sse_frame( + "content_block_start", + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + ), + _sse_frame( + "content_block_delta", + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}}, + ), + _sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}), + _sse_frame( + "message_delta", + { + "type": "message_delta", + "delta": { + "stop_reason": "refusal", + "stop_sequence": None, + "stop_details": {"type": "refusal", "category": "cyber", "explanation": "I can't help with that."}, + }, + "usage": {"output_tokens": 2}, + }, + ), + _sse_frame("message_stop", {"type": "message_stop"}), + ] + + +def _streamed_texts(chunks): + payloads = [json.loads(chunk.decode().split("data: ", 1)[1]) for chunk in chunks] + return [ + payload["delta"]["text"] + for payload in payloads + if payload["type"] == "content_block_delta" and payload["delta"]["type"] == "text_delta" + ] + + +@pytest.mark.asyncio +async def test_streaming_step_delivers_a_legacy_mask_of_a_streamed_claude_refusal(monkeypatch, caplog): + from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler + + chunks = _claude_refusal_sse_stream("Partial answer for bob@example.com") + + with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"): + result = await _run_legacy_streaming_step( + monkeypatch, _EmailMaskingLegacyHook(), chunks, translation=AnthropicMessagesHandler() + ) + + assert result.terminal_action == "allow" + assert not any("discarded" in record.getMessage() for record in caplog.records) + assert _streamed_texts(chunks) == ["Partial answer for [EMAIL]"] + + class _NoHooksGuardrail(CustomGuardrail): pass diff --git a/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py b/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py index 5627167432d..bc73ef8598a 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py +++ b/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py @@ -21,8 +21,9 @@ from litellm.responses.litellm_completion_transformation.session_handler import from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) -from litellm.types.llms.openai import ResponsesAPIStreamEvents +from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents from litellm.types.utils import Message, ModelResponse +from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE from tests.unit.proxy.conftest import httpx_transport @@ -1597,6 +1598,386 @@ async def test_previous_response_id_keeps_the_tools_an_earlier_codex_tool_search assert prisma_client.db.queries == 1 +_REFUSAL_EXPLANATION: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]["explanation"] +_ANSWER_BEFORE_REFUSAL: Final = {"type": "text", "text": "Here is the start."} + + +def _part_texts(item: BaseModel) -> list[tuple[str, str]]: + return [(part["type"], part.get("text") or part.get("refusal")) for part in item.model_dump()["content"]] + + +@pytest.mark.parametrize( + ("content", "expected_parts"), + [ + ((), [("refusal", _REFUSAL_EXPLANATION)]), + ( + (_ANSWER_BEFORE_REFUSAL,), + [("output_text", _ANSWER_BEFORE_REFUSAL["text"]), ("refusal", _REFUSAL_EXPLANATION)], + ), + ], + ids=["refusal_only", "answer_then_refusal"], +) +def test_responses_over_anthropic_return_the_refusal_explanation_as_a_refusal_part( + respx_mock: respx.MockRouter, content: tuple[Mapping[str, object], ...], expected_parts: list[tuple[str, str]] +) -> None: + respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response(200, json={**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": list(content)}) + ) + + response: Final = litellm.responses(model=_MODEL, input="Write the exploit.", api_key="mock_api_key") + + assert (response.status, response.incomplete_details.reason) == ("incomplete", "content_filter") + assert [_part_texts(item) for item in response.output] == [expected_parts] + + +def test_a_refused_responses_turn_sent_back_keeps_its_explanation_in_the_history( + respx_mock: respx.MockRouter, +) -> None: + route: Final = respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response(200, json={**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": []}) + ) + refused: Final = litellm.responses(model=_MODEL, input="Write the exploit.", api_key="mock_api_key") + + litellm.responses( + model=_MODEL, + input=[ + {"role": "user", "content": "Write the exploit."}, + *(item.model_dump(exclude_none=True) for item in refused.output), + {"role": "user", "content": "Then explain what SFT is."}, + ], + api_key="mock_api_key", + ) + + sent: Final = json.loads(route.calls.last.request.content) + assert [(message["role"], message["content"]) for message in sent["messages"][1:2]] == [ + ("assistant", [{"type": "text", "text": _REFUSAL_EXPLANATION}]) + ] + + +def _refusal_stream(*content: Mapping[str, object]) -> bytes: + text_events: Final = tuple( + chain.from_iterable( + ( + {"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}, + {"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": block["text"]}}, + {"type": "content_block_stop", "index": index}, + ) + for index, block in enumerate(content) + ) + ) + return _anthropic_sse( + { + "type": "message_start", + "message": {**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": [], "stop_reason": None, "stop_details": None}, + }, + *text_events, + { + "type": "message_delta", + "delta": { + "stop_reason": "refusal", + "stop_sequence": None, + "stop_details": ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"], + }, + "usage": {"output_tokens": 0}, + }, + {"type": "message_stop"}, + ) + + +_REFUSAL_PART_EVENTS: Final = ( + ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + ResponsesAPIStreamEvents.REFUSAL_DELTA, + ResponsesAPIStreamEvents.REFUSAL_DONE, + ResponsesAPIStreamEvents.CONTENT_PART_DONE, +) +_TEXT_PART_EVENTS: Final = ( + ResponsesAPIStreamEvents.CONTENT_PART_ADDED, + ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA, + ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE, + ResponsesAPIStreamEvents.CONTENT_PART_DONE, +) + + +@pytest.mark.parametrize( + ("content", "part_events", "expected_parts"), + [ + ((), _REFUSAL_PART_EVENTS, [("refusal", _REFUSAL_EXPLANATION)]), + ( + (_ANSWER_BEFORE_REFUSAL,), + (*_TEXT_PART_EVENTS, *_REFUSAL_PART_EVENTS), + [("output_text", _ANSWER_BEFORE_REFUSAL["text"]), ("refusal", _REFUSAL_EXPLANATION)], + ), + ], + ids=["refusal_only", "answer_then_refusal"], +) +def test_streamed_responses_over_anthropic_stream_the_refusal_explanation_as_a_refusal_part( + respx_mock: respx.MockRouter, + content: tuple[Mapping[str, object], ...], + part_events: tuple[ResponsesAPIStreamEvents, ...], + expected_parts: list[tuple[str, str]], +) -> None: + respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response( + 200, headers={"content-type": "text/event-stream"}, content=_refusal_stream(*content) + ) + ) + + events: Final = tuple( + litellm.responses(model=_MODEL, input="Write the exploit.", stream=True, api_key="mock_api_key") + ) + refusal_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE) + item_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE) + finished: Final = events[-1] + + assert tuple(event.type for event in events) == ( + ResponsesAPIStreamEvents.RESPONSE_CREATED, + ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS, + ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + *part_events, + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, + ) + assert (refusal_done.content_index, refusal_done.refusal) == (len(expected_parts) - 1, _REFUSAL_EXPLANATION) + assert _part_texts(item_done.item) == expected_parts + assert finished.response.incomplete_details.reason == "content_filter" + assert [_part_texts(item) for item in finished.response.output] == [expected_parts] + + +def test_streamed_responses_over_anthropic_give_a_refusal_after_thinking_its_own_message( + respx_mock: respx.MockRouter, +) -> None: + respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=_anthropic_sse( + { + "type": "message_start", + "message": { + **ANTHROPIC_DOCS_REFUSAL_MESSAGE, + "content": [], + "stop_reason": None, + "stop_details": None, + }, + }, + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Hmm."}}, + {"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig"}}, + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": { + "stop_reason": "refusal", + "stop_sequence": None, + "stop_details": ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"], + }, + "usage": {"output_tokens": 3}, + }, + {"type": "message_stop"}, + ), + ) + ) + + events: Final = tuple( + litellm.responses( + model=_MODEL, input="Write the exploit.", reasoning={"effort": "low"}, stream=True, api_key="mock_api_key" + ) + ) + refusal_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE) + finished: Final = events[-1] + + assert (refusal_done.output_index, refusal_done.content_index, refusal_done.refusal) == ( + 1, + 0, + _REFUSAL_EXPLANATION, + ) + assert [item.type for item in finished.response.output] == ["reasoning", "message"] + assert _part_texts(finished.response.output[1]) == [("refusal", _REFUSAL_EXPLANATION)] + + +_RAN_SEARCH: Final = { + "type": "server_tool_use", + "id": "srvtoolu_ran", + "name": "web_search", + "input": {"query": "tokyo"}, +} +_RAN_SEARCH_RESULT: Final = { + "type": "web_search_tool_result", + "tool_use_id": "srvtoolu_ran", + "content": [ + {"type": "web_search_result", "url": "https://tokyo.example", "title": "Tokyo", "encrypted_content": "e"} + ], +} +_THINKING_BEFORE_SEARCHES: Final = {"type": "thinking", "thinking": "Find each sister city in turn.", "signature": "sig"} +_TEXT_BEFORE_PENDING_SEARCH: Final = {"type": "text", "text": "Tokyo pairs with Paris. Checking Paris next."} +_PENDING_SEARCH: Final = { + "type": "server_tool_use", + "id": "srvtoolu_pending", + "name": "web_search", + "input": {"query": "next city"}, +} +_EXPECTED_PAUSED_OUTPUT: Final = [ + ("web_search_call", "completed", "tokyo"), + ("web_search_call", "in_progress", "next city"), +] + + +def _search_calls(output: Sequence[BaseModel]) -> list[tuple[str, str, str]]: + return [(item.type, item.status, item.action.query) for item in output if item.type == "web_search_call"] + + +def _reasoning_and_message_statuses(output: Sequence[BaseModel]) -> tuple[list[str], list[str]]: + return ( + [item.status for item in output if item.type == "reasoning"], + [item.status for item in output if item.type == "message"], + ) + + +def _last_item(output: Sequence[BaseModel]) -> tuple[str, str]: + return output[-1].type, output[-1].status + + +def _status_and_incomplete_details(response: ResponsesAPIResponse) -> tuple[str | None, Mapping[str, object] | None]: + details: Final = response.incomplete_details + return response.status, details.model_dump() if details is not None else None + + +_PAUSED_OR_CUT: Final = pytest.mark.parametrize( + ("stop_reason", "expected_status", "expected_incomplete_details"), + [ + ("pause_turn", "completed", None), + ("max_tokens", "incomplete", {"reason": "max_output_tokens"}), + ], + ids=["paused", "max_tokens"], +) + + +@_PAUSED_OR_CUT +def test_responses_over_anthropic_report_a_paused_turn_as_completed_with_its_pending_search( + respx_mock: respx.MockRouter, + stop_reason: str, + expected_status: str, + expected_incomplete_details: Mapping[str, object] | None, +) -> None: + respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response( + 200, + json={ + **ANTHROPIC_DOCS_REFUSAL_MESSAGE, + "content": [ + _THINKING_BEFORE_SEARCHES, + _RAN_SEARCH, + _RAN_SEARCH_RESULT, + _TEXT_BEFORE_PENDING_SEARCH, + _PENDING_SEARCH, + ], + "stop_reason": stop_reason, + "stop_details": None, + }, + ) + ) + + response: Final = litellm.responses( + model=_MODEL, input="Chain sister cities.", tools=[{"type": "web_search"}], api_key="mock_api_key" + ) + + assert _status_and_incomplete_details(response) == (expected_status, expected_incomplete_details) + assert _search_calls(response.output) == _EXPECTED_PAUSED_OUTPUT + assert _reasoning_and_message_statuses(response.output) == ([expected_status], [expected_status]) + assert _last_item(response.output) == ("web_search_call", "in_progress") + + +def _server_tool_use_events(index: int, block: Mapping[str, object]) -> tuple[Mapping[str, object], ...]: + return ( + {"type": "content_block_start", "index": index, "content_block": {**block, "input": {}}}, + { + "type": "content_block_delta", + "index": index, + "delta": {"type": "input_json_delta", "partial_json": json.dumps(block["input"])}, + }, + {"type": "content_block_stop", "index": index}, + ) + + +@_PAUSED_OR_CUT +def test_streamed_responses_over_anthropic_complete_a_paused_turn_with_its_pending_search_in_progress( + respx_mock: respx.MockRouter, + stop_reason: str, + expected_status: str, + expected_incomplete_details: Mapping[str, object] | None, +) -> None: + respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock( + return_value=httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=_anthropic_sse( + { + "type": "message_start", + "message": {**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": [], "stop_reason": None}, + }, + {"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}}, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "thinking_delta", "thinking": _THINKING_BEFORE_SEARCHES["thinking"]}, + }, + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "signature_delta", "signature": _THINKING_BEFORE_SEARCHES["signature"]}, + }, + {"type": "content_block_stop", "index": 0}, + *_server_tool_use_events(1, _RAN_SEARCH), + {"type": "content_block_start", "index": 2, "content_block": _RAN_SEARCH_RESULT}, + {"type": "content_block_stop", "index": 2}, + {"type": "content_block_start", "index": 3, "content_block": {"type": "text", "text": ""}}, + { + "type": "content_block_delta", + "index": 3, + "delta": {"type": "text_delta", "text": _TEXT_BEFORE_PENDING_SEARCH["text"]}, + }, + {"type": "content_block_stop", "index": 3}, + *_server_tool_use_events(4, _PENDING_SEARCH), + { + "type": "message_delta", + "delta": {"stop_reason": stop_reason, "stop_sequence": None}, + "usage": {"output_tokens": 5}, + }, + {"type": "message_stop"}, + ), + ) + ) + + events: Final = tuple( + litellm.responses( + model=_MODEL, + input="Chain sister cities.", + tools=[{"type": "web_search"}], + stream=True, + api_key="mock_api_key", + ) + ) + pending_item_id: Final = f"ws_{_PENDING_SEARCH['id']}" + pending_events: Final = [ + event.type + for event in events + if getattr(event, "item_id", None) == pending_item_id + or getattr(getattr(event, "item", None), "id", None) == pending_item_id + ] + finished: Final = events[-1] + + assert pending_events == [ + ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED, + ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS, + ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, + ] + assert finished.type == f"response.{expected_status}" + assert _status_and_incomplete_details(finished.response) == (expected_status, expected_incomplete_details) + assert _search_calls(finished.response.output) == _EXPECTED_PAUSED_OUTPUT + assert _reasoning_and_message_statuses(finished.response.output) == ([expected_status], [expected_status]) + assert _last_item(finished.response.output) == ("web_search_call", "in_progress") + + async def _codex_follow_up_body(**kwargs: object) -> Mapping[str, object]: anthropic: Final = _RecordingRawAnthropicMessages() with patch("litellm.proxy.proxy_server.prisma_client", _FakePrismaClient([_SEARCH_TURN, _CALENDAR_TURN])): diff --git a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py index b1047a345d7..b4b4bb9bd07 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -4339,6 +4339,7 @@ class TestEnsureOutputItemContentPartAdded: iterator = LiteLLMCompletionStreamingIterator.__new__(LiteLLMCompletionStreamingIterator) iterator.sent_output_item_added_event = False + iterator.sent_message_item_added_event = False iterator.sent_content_part_added_event = False iterator._sequence_number = 0 iterator._cached_item_id = None @@ -6816,6 +6817,51 @@ def _reasoning_item(item_id: str, thinking: Mapping[str, str]) -> Mapping[str, o return {"type": "reasoning", "id": item_id, "summary": [], "encrypted_content": encode_thinking_blocks([thinking])} +def _hosted_search(search_id: str, status: str) -> Mapping[str, object]: + return { + "type": "web_search_call", + "id": search_id, + "status": status, + "action": {"type": "search", "query": search_id, "queries": [search_id], "sources": []}, + } + + +@pytest.mark.parametrize( + ("last_search_status", "replayed_thinking"), + [ + pytest.param("in_progress", None, id="paused_turn"), + pytest.param("completed", [_ANSWERED_THINKING], id="finished_turn"), + ], +) +def test_a_new_message_after_a_paused_search_turn_replays_only_that_turn_without_its_thinking( + last_search_status: str, replayed_thinking: list[Mapping[str, str]] | None +) -> None: + tokyo_thinking: Final = {"type": "thinking", "thinking": "Need Tokyo weather.", "signature": "sig_tokyo"} + messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( + model="anthropic/claude-sonnet-5-5", + input=cast( + ResponseInputParam, + [ + {"role": "user", "content": "Plan a Tokyo trip."}, + _reasoning_item("rs_tokyo", tokyo_thinking), + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Tokyo is sunny."}]}, + _hosted_search("ws_srvtoolu_01Tokyo", "completed"), + {"role": "user", "content": "Now Osaka."}, + _reasoning_item("rs_osaka", _ANSWERED_THINKING), + _CHECKING_OSAKA, + _hosted_search("ws_srvtoolu_02Osaka", "completed"), + _hosted_search("ws_srvtoolu_03Osaka", last_search_status), + {"role": "user", "content": "Please continue."}, + ], + ), + responses_api_request=ResponsesAPIOptionalRequestParams(), + )["messages"] + + assert [message["role"] for message in messages] == ["user", "assistant", "user", "assistant", "user"] + assert messages[1].get("thinking_blocks") == [tokyo_thinking] + assert messages[3].get("thinking_blocks") == replayed_thinking + + def test_summary_replay_keeps_only_the_reasoning_of_the_turn_whose_call_a_later_output_answers() -> None: tokyo_thinking: Final = {"type": "thinking", "thinking": "Need Tokyo weather.", "signature": "sig_tokyo"} messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( diff --git a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index 86773dc7ddf..684f5a89e80 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -1988,3 +1988,41 @@ def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_b partial.response.usage.input_tokens, partial.response.usage.output_tokens, ), "the bill is priced on the partial's own usage" + + +def _refusal_chunk(refusal: str, finish_reason: str | None = None) -> ModelResponseStream: + return ModelResponseStream( + id=CHAT_COMPLETION_ID, + created=1748575031, + model="gpt-5.5", + object="chat.completion.chunk", + choices=[ + StreamingChoices( + index=0, + delta=Delta(role="assistant", content=None, refusal=refusal), + finish_reason=finish_reason, + ) + ], + ) + + +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.asyncio +async def test_streamed_refusal_deltas_become_one_refusal_part_of_the_message(sync_mode: bool): + iterator: Final = _build_iterator( + [_refusal_chunk("I'm sorry, "), _refusal_chunk("I can't help with that."), _chunk("", finish_reason="stop")] + ) + + events: Final = await _collect_events(iterator, sync_mode) + + parts_added: Final = [ + event.model_dump()["part"] for event in events if event.type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED + ] + refusal_done: Final = [event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE] + message_done: Final = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE] + finished: Final = events[-1] + expected_content: Final = [{"type": "refusal", "refusal": "I'm sorry, I can't help with that."}] + assert parts_added == [{"type": "refusal", "refusal": ""}] + assert [event.refusal for event in refusal_done] == [expected_content[0]["refusal"]] + assert [item.model_dump()["content"] for item in message_done] == [expected_content] + assert [item.model_dump()["content"] for item in finished.response.output] == [expected_content] diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 02b8a80e128..cff0611033f 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -33337,7 +33337,7 @@ export interface components { */ GenericResponseOutputItem: { /** Content */ - content: components["schemas"]["OutputText"][]; + content: (components["schemas"]["OutputText"] | components["schemas"]["ResponseOutputRefusal"])[]; /** Id */ id: string; /** Phase */