scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-11 22:51:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 124

fix(anthropic): harden refusal translation

Browse source
This commit is contained in:
Atharva-Kanherkar 2026-09-04 17:02:07 +05:30
parent 200e2901d6
commit 0b34abe8fe
10 changed files with 397 additions and 102 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

39
litellm/llms/anthropic/experimental_pass_through/adapters/streaming_iterator.py
View file

@ -4,13 +4,14 @@ import copy
import json
import traceback
from collections import deque
from collections.abc import AsyncIterator, Iterator, Sequence
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from typing import (
TYPE_CHECKING,
Any,
Final,
Literal,
Protocol,
cast,
get_args,
)
@ -305,6 +306,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
# Synthesized compaction block from compact_20260112 polyfill (streaming).
self.compaction_block = compaction_block
self.iterations_usage = iterations_usage
self._refusal_text_parts: list[str] = []
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
@ -572,6 +574,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)
# Check if this is a usage chunk and we have a held stop_reason chunk
if will_merge_into_held:
@ -806,6 +809,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)
# Check if this is a usage chunk and we have a held stop_reason chunk
if will_merge_into_held:
@ -993,6 +997,31 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
def _increment_content_block_index(self):
self.current_content_block_index += 1
def _with_refusal_stop_details(
self,
processed_chunk: ContentBlockDelta | MessageBlockDelta,
) -> ContentBlockDelta | MessageBlockDelta:
if processed_chunk.get("type") != "message_delta" or not self._refusal_text_parts:
return processed_chunk
delta: Final = cast(Mapping[str, object], processed_chunk["delta"])
if delta.get("stop_reason") == "max_tokens":
return processed_chunk
return cast(
ContentBlockDelta | MessageBlockDelta,
{
**processed_chunk,
"delta": {
**delta,
"stop_reason": "refusal",
"stop_details": {
"type": "refusal",
"category": None,
"explanation": "".join(self._refusal_text_parts),
},
},
},
)
@staticmethod
def _delta_has_content(processed_chunk: dict[str, Any]) -> bool:
"""Return True if a translated chunk carries a non-empty
@ -1044,6 +1073,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
return False
if getattr(delta, "content", None):
return False
if getattr(delta, "refusal", None):
return False
if getattr(delta, "reasoning_content", None):
return False
# thinking_blocks whose entries are all empty AND unsigned must not
@ -1069,8 +1100,10 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
"""
from .transformation import LiteLLMAnthropicMessagesAdapter
# Example logic - customize based on your needs:
# If chunk indicates a tool call
refusal_text: Final = LiteLLMAnthropicMessagesAdapter._refusal_text(chunk.choices[0].delta)
if refusal_text is not None:
self._refusal_text_parts.append(refusal_text)
if chunk.choices[0].finish_reason is not None:
return False

37
litellm/llms/anthropic/experimental_pass_through/adapters/transformation.py
View file

@ -307,6 +307,17 @@ class LiteLLMAnthropicMessagesAdapter:
def __init__(self):
pass
@staticmethod
def _refusal_text(message_or_delta: object) -> str | None:
refusal: Final = getattr(message_or_delta, "refusal", None)
if isinstance(refusal, str):
return refusal
provider_specific_fields: Final = getattr(message_or_delta, "provider_specific_fields", None)
if isinstance(provider_specific_fields, Mapping):
provider_refusal: Final = provider_specific_fields.get("refusal")
return provider_refusal if isinstance(provider_refusal, str) else None
return None
### FOR [BETA] `/v1/messages` endpoint support
def _extract_signature_from_tool_call(self, tool_call: object) -> str | None:
@ -1324,6 +1335,8 @@ class LiteLLMAnthropicMessagesAdapter:
new_content.append(
AnthropicResponseContentBlockText(type="text", text=choice.message.content).model_dump()
)
if (refusal_text := self._refusal_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:
for tool_call in choice.message.tool_calls:
@ -1482,14 +1495,23 @@ class LiteLLMAnthropicMessagesAdapter:
choices=response.choices,
tool_name_mapping=tool_name_mapping,
)
refusal_text: Final = next(
(text for choice in response.choices if (text := self._refusal_text(choice.message)) is not None),
None,
)
if polyfill_result is not None and polyfill_result.compaction_block is not None:
anthropic_content.insert(0, polyfill_result.compaction_block)
## extract finish reason
anthropic_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic(
translated_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic(
openai_finish_reason=response.choices[0].finish_reason
)
anthropic_finish_reason: Final = (
"refusal"
if refusal_text is not None and translated_finish_reason != "max_tokens"
else translated_finish_reason
)
# extract usage
usage: Final[Usage] = getattr(response, "usage")
anthropic_usage: Final = self._translate_openai_usage_to_anthropic_usage(usage)
@ -1511,6 +1533,15 @@ class LiteLLMAnthropicMessagesAdapter:
usage=anthropic_usage,
content=anthropic_content,
stop_reason=anthropic_finish_reason,
stop_details=(
{
"type": "refusal",
"category": None,
"explanation": refusal_text,
}
if anthropic_finish_reason == "refusal"
else None
),
)
applied_edits: Final = polyfill_result.applied_edits_for_response() if polyfill_result else None
@ -1551,7 +1582,9 @@ 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:
elif (choice.delta.content is not None and len(choice.delta.content) > 0) or self._refusal_text(
choice.delta
) is not None:
return "text", TextBlock(type="text", text="")
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
thinking_blocks = choice.delta.thinking_blocks or []

114
litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py
View file

@ -1,5 +1,6 @@
# What is this?
## Translates OpenAI call to Anthropic `/v1/messages` format
import asyncio
import json
import traceback
from collections import deque
@ -49,6 +50,8 @@ class AnthropicResponsesStreamWrapper:
self._sent_message_start = False
self._sent_message_stop = False
self._chunk_queue: deque = deque()
self._refusal_text_parts: list[str] = []
self._sync_responses_iterator: Any = None
def _make_message_start(self) -> dict[str, Any]:
return {
@ -111,7 +114,7 @@ class AnthropicResponsesStreamWrapper:
item_type: Final = getattr(item, "type", None) or (item.get("type") if isinstance(item, dict) else None)
item_id = getattr(item, "id", None) or (item.get("id") if isinstance(item, dict) else None)
if item_type in ("message", "refusal"):
if item_type == "message":
self._open_block(item_id, {"type": "text", "text": ""})
elif item_type == "function_call":
call_id: Final = (
@ -131,8 +134,14 @@ class AnthropicResponsesStreamWrapper:
)
return
if event_type == "response.refusal.delta":
delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
if isinstance(delta, str):
self._refusal_text_parts.append(delta)
return
# ---- text delta ----
if event_type in ("response.output_text.delta", "response.refusal.delta"):
if event_type == "response.output_text.delta":
item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None)
delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "")
block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index
@ -215,52 +224,53 @@ class AnthropicResponsesStreamWrapper:
response_obj: Final = getattr(event, "response", None) or (
event.get("response") if isinstance(event, dict) else None
)
stop_reason = "end_turn"
anthropic_usage: AnthropicUsage = AnthropicUsage(input_tokens=0, output_tokens=0)
if response_obj is not None:
status: Final = getattr(response_obj, "status", None)
if status == "incomplete":
stop_reason = "max_tokens"
anthropic_usage = (
LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(
getattr(response_obj, "usage", None)
)
output: Final = (getattr(response_obj, "output", None) or ()) if response_obj is not None else ()
refusal_text: Final = LiteLLMAnthropicToResponsesAPIAdapter._refusal_text_from_output(output) or (
"".join(self._refusal_text_parts) or None
)
status: Final = getattr(response_obj, "status", None) if response_obj is not None else None
has_tool_call: Final = any(
getattr(item, "type", None) == "function_call"
or (isinstance(item, dict) and item.get("type") == "function_call")
for item in output
)
stop_reason: Final = (
"max_tokens"
if status == "incomplete"
else "refusal"
if refusal_text is not None
else "tool_use"
if has_tool_call
else "end_turn"
)
anthropic_usage: Final[AnthropicUsage] = (
LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(
getattr(response_obj, "usage", None)
)
if response_obj is not None
else AnthropicUsage(input_tokens=0, output_tokens=0)
)
# Check if tool_use was in the output to override stop_reason
if response_obj is not None:
output: Final = getattr(response_obj, "output", []) or []
for out_item in output:
out_type = getattr(out_item, "type", None) or (
out_item.get("type") if isinstance(out_item, dict) else None
)
if out_type == "function_call":
stop_reason = "tool_use"
break
elif out_type == "refusal":
stop_reason = "refusal"
break
elif out_type == "message":
content_parts = getattr(out_item, "content", ()) or (
out_item.get("content") or () if isinstance(out_item, dict) else ()
)
for part in content_parts:
part_type = getattr(part, "type", None) or (
part.get("type") if isinstance(part, dict) else None
)
if (
part_type == "refusal"
or hasattr(part, "refusal")
or (isinstance(part, dict) and "refusal" in part)
):
stop_reason = "refusal"
break
message_delta_payload: Final = {
"stop_reason": stop_reason,
"stop_sequence": None,
**(
{
"stop_details": {
"type": "refusal",
"category": None,
"explanation": refusal_text,
}
}
if stop_reason == "refusal"
else {}
),
}
self._chunk_queue.append(
{
"type": "message_delta",
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
"delta": message_delta_payload,
"usage": dict(anthropic_usage),
}
)
@ -284,10 +294,22 @@ class AnthropicResponsesStreamWrapper:
# Consume the upstream stream
try:
async for event in self.responses_stream:
self._process_event(event)
if self._chunk_queue:
return self._chunk_queue.popleft()
if hasattr(self.responses_stream, "__aiter__"):
async for event in self.responses_stream:
self._process_event(event)
if self._chunk_queue:
return self._chunk_queue.popleft()
else:
if self._sync_responses_iterator is None:
self._sync_responses_iterator = iter(self.responses_stream)
missing: Final = object()
while True:
event = await asyncio.to_thread(next, self._sync_responses_iterator, missing)
if event is missing:
break
self._process_event(event)
if self._chunk_queue:
return self._chunk_queue.popleft()
except StopAsyncIteration:
pass
except Exception as e:

58
litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py
View file

@ -6,7 +6,7 @@ path used for OpenAI and Azure models.
"""
import json
from collections.abc import Iterable, Mapping
from collections.abc import Iterable, Mapping, Sequence
from itertools import groupby
from typing import Any, Final, cast
@ -69,6 +69,38 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
chat_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(raw_usage)
return LiteLLMAnthropicMessagesAdapter._translate_openai_usage_to_anthropic_usage(chat_usage)
@staticmethod
def _refusal_text_from_output(output: Iterable[object]) -> str | None:
from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal
def refusal_text_from_item(item: object) -> str | None:
if isinstance(item, ResponseOutputMessage):
return next(
(part.refusal for part in item.content if isinstance(part, ResponseOutputRefusal)),
None,
)
if not isinstance(item, Mapping):
return None
item_mapping: Final = cast(Mapping[str, object], item)
raw_parts: Final = item_mapping.get("content")
if item_mapping.get("type") != "message" or not isinstance(raw_parts, Sequence):
return None
return next(
(
refusal
for part in cast(Sequence[object], raw_parts)
if isinstance(part, Mapping)
and cast(Mapping[str, object], part).get("type") == "refusal"
and isinstance((refusal := cast(Mapping[str, object], part).get("refusal")), str)
),
None,
)
return next(
(text for item in output if (text := refusal_text_from_item(item)) is not None),
None,
)
# ------------------------------------------------------------------ #
# Request translation: Anthropic -> Responses API #
# ------------------------------------------------------------------ #
@ -624,6 +656,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
content: Final[list[dict[str, object]]] = []
stop_reason: AnthropicFinishReason = "end_turn"
refusal_text: Final = self._refusal_text_from_output(cast(Iterable[object], response.output))
for item in response.output:
if isinstance(item, ResponseReasoningItem):
@ -636,13 +669,12 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
content.append(
AnthropicResponseContentBlockText(type="text", text=getattr(part, "text", "")).model_dump()
)
elif part_type == "refusal" or hasattr(part, "refusal"):
elif part_type == "refusal":
content.append(
AnthropicResponseContentBlockText(
type="text", text=getattr(part, "refusal", "") or ""
).model_dump()
)
stop_reason = "refusal"
elif isinstance(item, ResponseFunctionToolCall):
try:
@ -671,13 +703,12 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
type="text", text=part.get("text", "")
).model_dump()
)
elif part_type == "refusal" or "refusal" in part:
elif part_type == "refusal":
content.append(
AnthropicResponseContentBlockText(
type="text", text=part.get("refusal", "") or ""
).model_dump()
)
stop_reason = "refusal"
elif item_type == "reasoning":
content.extend(
self._thinking_blocks_from_reasoning_item(
@ -698,14 +729,10 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
).model_dump()
)
stop_reason = "tool_use"
elif item_type == "refusal":
refusal_text = item.get("refusal") or item.get("text", "") or ""
content.append(AnthropicResponseContentBlockText(type="text", text=refusal_text).model_dump())
stop_reason = "refusal"
# status -> stop_reason override
if response.status == "incomplete":
stop_reason = "max_tokens"
elif refusal_text is not None:
stop_reason = "refusal"
anthropic_usage: Final = self.translate_responses_api_usage_to_anthropic_usage(response.usage)
@ -718,4 +745,13 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
usage=anthropic_usage,
content=content,
stop_reason=stop_reason,
stop_details=(
{
"type": "refusal",
"category": None,
"explanation": refusal_text,
}
if stop_reason == "refusal"
else None
),
)

7
litellm/types/llms/anthropic.py
View file

@ -520,8 +520,15 @@ ContentBlockContentBlockDict = ToolUseBlock | TextBlock | ChatCompletionThinking
ContentBlockStart = ContentBlockStartToolUse | ContentBlockStartText
class AnthropicStopDetails(TypedDict, total=False):
type: ReadOnly[Literal["refusal"]]
category: ReadOnly[str | None]
explanation: ReadOnly[str | None]
class MessageDelta(TypedDict, total=False):
stop_reason: str | None
stop_details: AnthropicStopDetails
class ServerToolUsage(TypedDict, total=False):

11
litellm/types/llms/anthropic_messages/anthropic_response.py
View file

@ -5,6 +5,7 @@ from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.types.llms.anthropic import (
AnthropicResponseContentBlockText,
AnthropicResponseContentBlockToolUse,
AnthropicStopDetails,
ContextManagementResponse,
ServerToolUsage,
)
@ -78,16 +79,6 @@ class AnthropicUsage(TypedDict, total=False):
server_tool_use: NotRequired[ReadOnly[ServerToolUsage]]
class AnthropicStopDetails(TypedDict, total=False):
"""
Safeguard verdict accompanying a `stop_reason: "refusal"` response:
https://platform.claude.com/docs/en/build-with-claude/refusals-and-fallback
"""
category: ReadOnly[str | None]
explanation: ReadOnly[str | None]
class AnthropicMessagesResponse(TypedDict, total=False):
"""
Anthropic Messages API Response: https://docs.anthropic.com/en/api/messages

45
tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py
View file

@ -40,6 +40,51 @@ from litellm.types.utils import (
)
def test_translate_chat_refusal_to_anthropic_response():
response = ModelResponse(
id="chatcmpl-refusal",
model="openai-model",
choices=[
Choices(
index=0,
finish_reason="stop",
message=Message(content=None, role="assistant", refusal="I cannot fulfill this request."),
)
],
usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2),
)
result = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response)
assert result["content"] == [{"type": "text", "text": "I cannot fulfill this request."}]
assert result["stop_reason"] == "refusal"
assert result.get("stop_details") == {
"type": "refusal",
"category": None,
"explanation": "I cannot fulfill this request.",
}
def test_translate_chat_length_takes_precedence_over_refusal():
response = ModelResponse(
id="chatcmpl-partial-refusal",
model="openai-model",
choices=[
Choices(
index=0,
finish_reason="length",
message=Message(content=None, role="assistant", refusal="Partial refusal"),
)
],
usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2),
)
result = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response)
assert result["stop_reason"] == "max_tokens"
assert result.get("stop_details") is None
def test_translate_streaming_openai_chunk_to_anthropic_content_block():
choices = [
StreamingChoices(

87
tests/test_litellm/llms/anthropic/experimental_pass_through/adapters/test_streaming_iterator_first_delta.py
View file

@ -108,6 +108,93 @@ def _text_deltas(events: List[dict]) -> List[str]:
]
def test_streaming_chat_refusal_emits_only_refusal_stop_details():
chunks = [
_make_chunk(Delta(content=None, refusal="I cannot fulfill this request.")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="openai-model")
events = _drain_sync(wrapper)
assert _text_deltas(events) == []
message_delta = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"] == {
"stop_reason": "refusal",
"stop_details": {
"type": "refusal",
"category": None,
"explanation": "I cannot fulfill this request.",
},
}
@pytest.mark.asyncio
async def test_streaming_chat_refusal_emits_only_refusal_stop_details_async():
chunks = [
_make_chunk(Delta(content=None, refusal="I cannot fulfill this request.")),
_make_chunk(Delta(content=None), finish_reason="stop"),
]
wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="openai-model")
events = await _drain_async(wrapper)
assert _text_deltas(events) == []
message_delta = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "refusal"
assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request."
def test_streaming_chat_combined_refusal_and_finish_reason_is_preserved():
chunks = [
_make_chunk(
Delta(content=None, refusal="I cannot fulfill this request."),
finish_reason="stop",
)
]
wrapper = AnthropicStreamWrapper(completion_stream=iter(chunks), model="openai-model")
events = _drain_sync(wrapper)
message_delta = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "refusal"
assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request."
@pytest.mark.asyncio
async def test_streaming_chat_combined_refusal_and_finish_reason_is_preserved_async():
chunks = [
_make_chunk(
Delta(content=None, refusal="I cannot fulfill this request."),
finish_reason="stop",
)
]
wrapper = AnthropicStreamWrapper(completion_stream=_AsyncStream(chunks), model="openai-model")
events = await _drain_async(wrapper)
message_delta = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "refusal"
assert message_delta["delta"]["stop_details"]["explanation"] == "I cannot fulfill this request."
@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.asyncio
async def test_streaming_chat_length_takes_precedence_over_refusal(async_mode: bool):
chunks = [
_make_chunk(Delta(content=None, refusal="Partial refusal")),
_make_chunk(Delta(content=None), finish_reason="length"),
]
stream = _AsyncStream(chunks) if async_mode else iter(chunks)
wrapper = AnthropicStreamWrapper(completion_stream=stream, model="openai-model")
events = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper)
message_delta = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "max_tokens"
assert "stop_details" not in message_delta["delta"]
def _input_json_deltas(events: List[dict]) -> List[str]:
return [
e["delta"]["partial_json"]

60
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py
View file

@ -34,6 +34,14 @@ def _drain_async(events: list) -> list:
return asyncio.run(_run())
def _drain_sync_upstream(events: list) -> list:
async def _run() -> list:
wrapper = AnthropicResponsesStreamWrapper(responses_stream=iter(events), model="m")
return [chunk async for chunk in wrapper]
return asyncio.run(_run())
class TestMessageStartEmittedExactlyOnce:
"""The ``__anext__`` fallback emits ``message_start`` before consuming the
stream, so ``_process_event`` must not emit a second one when
@ -55,6 +63,15 @@ class TestMessageStartEmittedExactlyOnce:
chunks = _drain_async([{"type": "response.created"}])
assert chunks[0]["type"] == "message_start"
def test_sync_upstream_iterator_is_consumed(self):
chunks = _drain_sync_upstream(
[
{"type": "response.created"},
{"type": "response.output_text.delta", "item_id": "m1", "delta": "hi"},
]
)
assert any(chunk.get("delta", {}).get("text") == "hi" for chunk in chunks)
class TestProcessEventResponseCreatedGuard:
"""``_process_event`` must emit ``message_start`` exactly once even if
@ -311,17 +328,37 @@ class TestResponseCompletedUsage:
class TestRefusalStreamEvents:
def test_refusal_delta_emits_text_delta(self):
def test_refusal_event_sequence_emits_only_stop_details(self):
response = SimpleNamespace(
status="completed",
output=[{"type": "message", "content": [{"type": "refusal", "refusal": "I cannot fulfill this."}]}],
usage=None,
)
chunks = _process_all(
[
{"type": "response.created"},
{"type": "response.refusal.delta", "item_id": "ref_1", "delta": "I cannot fulfill this."},
{"type": "response.output_item.added", "item": {"type": "message", "id": "msg_1"}},
{"type": "response.refusal.delta", "item_id": "msg_1", "delta": "I cannot fulfill this."},
{"type": "response.output_item.done", "item": {"type": "message", "id": "msg_1"}},
{"type": "response.completed", "response": response},
]
)
assert any(
c.get("type") == "content_block_delta" and c.get("delta", {}).get("text") == "I cannot fulfill this."
for c in chunks
)
assert [chunk["type"] for chunk in chunks] == [
"message_start",
"content_block_start",
"content_block_stop",
"message_delta",
"message_stop",
]
assert chunks[3]["delta"] == {
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": {
"type": "refusal",
"category": None,
"explanation": "I cannot fulfill this.",
},
}
def test_response_completed_with_refusal_sets_stop_reason_refusal(self):
response = SimpleNamespace(
@ -333,12 +370,13 @@ class TestRefusalStreamEvents:
message_delta = next(c for c in chunks if c["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "refusal"
def test_response_completed_with_standalone_refusal_item_sets_stop_reason_refusal(self):
def test_incomplete_status_takes_precedence_over_refusal(self):
response = SimpleNamespace(
status="completed",
output=[{"type": "refusal", "refusal": "Standalone refusal"}],
status="incomplete",
output=[{"type": "message", "content": [{"type": "refusal", "refusal": "Partial refusal"}]}],
usage=None,
)
chunks = _process_all([{"type": "response.completed", "response": response}])
chunks = _process_all([{"type": "response.incomplete", "response": response}])
message_delta = next(c for c in chunks if c["type"] == "message_delta")
assert message_delta["delta"]["stop_reason"] == "refusal"
assert message_delta["delta"]["stop_reason"] == "max_tokens"
assert "stop_details" not in message_delta["delta"]

41
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py
View file

@ -1205,16 +1205,16 @@ def _make_output_message(texts: List[str]) -> MagicMock:
return msg
def _make_refusal_message(refusal_text: str) -> MagicMock:
from openai.types.responses import ResponseOutputMessage
def _make_refusal_message(refusal_text: str):
from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal
part = MagicMock()
part.type = "refusal"
part.refusal = refusal_text
msg = MagicMock(spec=ResponseOutputMessage)
msg.content = [part]
return msg
return ResponseOutputMessage(
id="msg_refusal",
content=[ResponseOutputRefusal(type="refusal", refusal=refusal_text)],
role="assistant",
status="completed",
type="message",
)
def _make_function_call_item(call_id: str, name: str, arguments: str) -> MagicMock:
@ -1296,6 +1296,11 @@ class TestTranslateResponse:
assert result["content"][0]["type"] == "text"
assert result["content"][0]["text"] == "I cannot fulfill this request."
assert result["stop_reason"] == "refusal"
assert result.get("stop_details") == {
"type": "refusal",
"category": None,
"explanation": "I cannot fulfill this request.",
}
def test_dict_refusal_part_in_message_becomes_text_block(self):
output_item = {
@ -1308,18 +1313,16 @@ class TestTranslateResponse:
assert result["content"][0]["type"] == "text"
assert result["content"][0]["text"] == "Refused by policy"
assert result["stop_reason"] == "refusal"
assert result.get("stop_details", {}).get("explanation") == "Refused by policy"
def test_dict_refusal_item_becomes_text_block(self):
output_item = {
"type": "refusal",
"refusal": "Standalone refusal",
}
response = _make_mock_response(output=[output_item])
def test_incomplete_status_takes_precedence_over_refusal(self):
response = _make_mock_response(
output=[_make_refusal_message("Partial refusal")],
status="incomplete",
)
result: Any = _ADAPTER.translate_response(response)
assert len(result["content"]) == 1
assert result["content"][0]["type"] == "text"
assert result["content"][0]["text"] == "Standalone refusal"
assert result["stop_reason"] == "refusal"
assert result["stop_reason"] == "max_tokens"
assert result.get("stop_details") is None
def test_incomplete_status_sets_max_tokens(self):
"""status='incomplete' overrides stop_reason to 'max_tokens'."""