diff --git a/litellm/litellm_core_utils/redact_messages.py b/litellm/litellm_core_utils/redact_messages.py index ac5c0720eff..58dab22078e 100644 --- a/litellm/litellm_core_utils/redact_messages.py +++ b/litellm/litellm_core_utils/redact_messages.py @@ -212,6 +212,7 @@ def _redact_choice_content(choice): choice.message.content = REDACTED_BY_LITELLM if getattr(choice.message, "reasoning_content", None) is not None: choice.message.reasoning_content = REDACTED_BY_LITELLM + _redact_field_when_present(choice.message, "refusal") if hasattr(choice.message, "thinking_blocks"): choice.message.thinking_blocks = None _drop_field_when_present(choice.message, "reasoning_items") @@ -223,6 +224,7 @@ def _redact_choice_content(choice): choice.delta.content = REDACTED_BY_LITELLM if getattr(choice.delta, "reasoning_content", None) is not None: choice.delta.reasoning_content = REDACTED_BY_LITELLM + _redact_field_when_present(choice.delta, "refusal") if hasattr(choice.delta, "thinking_blocks"): choice.delta.thinking_blocks = None _drop_field_when_present(choice.delta, "reasoning_items") @@ -479,7 +481,7 @@ def _redact_model_response_dict_choices(choices: Iterable[object], redacted_str: if ( body := next((choice[key] for key in ("message", "delta") if isinstance(choice.get(key), dict)), None) ) is not None: - for field in ("content", "reasoning_content"): + for field in ("content", "reasoning_content", "refusal"): if body.get(field) is not None: body[field] = redacted_str for field in ("thinking_blocks", "reasoning_items", "audio"): diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index c87357122ce..a3982f87a4a 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -1556,6 +1556,11 @@ def _chunk_refusal(chunk: object) -> str: return refusal if isinstance(refusal, str) else "" +def _message_refusal(choice: object) -> str: + refusal: Final = _field_of(_field_of(choice, "message"), "refusal") + return refusal if isinstance(refusal, str) else "" + + _NON_TEXT_DELTA_FIELDS: Final = ( "tool_calls", "function_call", @@ -1906,7 +1911,7 @@ def _assembled_stream_response( for field, value in _assembled_message_fields(processor, choice_chunks): choice.message[field] = value - completion_output = get_content_from_model_response(response) + completion_output = get_content_from_model_response(response) + "".join(map(_message_refusal, response.choices)) reasoning_tokens: Final = processor.count_reasoning_tokens(response) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index aa6ce6ea99e..ae53e282e77 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -1129,6 +1129,7 @@ class CustomStreamWrapper: and model_response.choices[0].delta.role is not None ) or (getattr(model_response.choices[0].delta, "reasoning_items", None) is not None) + or bool(getattr(model_response.choices[0].delta, "refusal", None)) ): return True else: @@ -2800,6 +2801,7 @@ class CustomStreamWrapper: "audio", "images", "annotations", + "refusal", ) @classmethod diff --git a/litellm/router_python_core.py b/litellm/router_python_core.py index 8ca2afb9a99..d16ad843230 100644 --- a/litellm/router_python_core.py +++ b/litellm/router_python_core.py @@ -686,6 +686,7 @@ def _stream_chunks_have_generated_content(chunks: Sequence[ModelResponseStream]) or delta.get("audio") or delta.get("images") or delta.get("annotations") + or getattr(delta, "refusal", None) ): return True return False diff --git a/tests/unit/litellm_core_utils/test_redact_messages.py b/tests/unit/litellm_core_utils/test_redact_messages.py index 491ea5315ff..8b198a96178 100644 --- a/tests/unit/litellm_core_utils/test_redact_messages.py +++ b/tests/unit/litellm_core_utils/test_redact_messages.py @@ -20,6 +20,7 @@ from unittest.mock import patch import httpx import pytest import pytest_asyncio +from openai import AsyncOpenAI from pydantic import JsonValue import litellm @@ -1798,6 +1799,72 @@ async def test_responses_redaction_removes_request_instructions_from_logged_kwar assert SECRET not in json.dumps({key: kwargs.get(key) for key in logged_keys}, default=str) +_REFUSAL_PIECES: Final = ("I can't share ", SECRET) + + +def _chat_chunk(delta: Mapping[str, object], finish_reason: str | None) -> str: + chunk: Final = { + "id": "chatcmpl-refusal", + "object": "chat.completion.chunk", + "created": 0, + "model": "gpt-4.1", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], + } + return f"data: {json.dumps(chunk)}\n\n" + + +def _openai_refusal_reply(request: httpx.Request) -> httpx.Response: + if json.loads(request.content).get("stream"): + deltas: Final = ({"role": "assistant", "refusal": ""}, *({"refusal": piece} for piece in _REFUSAL_PIECES)) + sse: Final = "".join(_chat_chunk(delta, None) for delta in deltas) + _chat_chunk({}, "stop") + "data: [DONE]\n\n" + return httpx.Response(200, request=request, headers={"content-type": "text/event-stream"}, content=sse) + message: Final = {"role": "assistant", "content": None, "refusal": "".join(_REFUSAL_PIECES)} + return httpx.Response( + 200, + request=request, + json={ + "id": "chatcmpl-refusal", + "object": "chat.completion", + "created": 0, + "model": "gpt-4.1", + "choices": [{"index": 0, "message": message, "finish_reason": "stop"}], + }, + ) + + +async def _caller_refusal(response: object) -> str | None: + if isinstance(response, litellm.CustomStreamWrapper): + return "".join([getattr(chunk.choices[0].delta, "refusal", None) or "" async for chunk in response]) + assert isinstance(response, ModelResponse) + return getattr(response.choices[0].message, "refusal", None) # pyright: ignore[reportAttributeAccessIssue] # choices is a Choices | StreamingChoices union + + +@pytest.mark.usefixtures("drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") +@pytest.mark.asyncio +@pytest.mark.parametrize("stream", [True, False]) +async def test_redaction_hides_a_refusal_from_loggers_but_not_from_the_caller( + stream: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + test_custom_logger: Final = TestCustomLogger() + monkeypatch.setattr(litellm, "turn_off_message_logging", True) + monkeypatch.setattr(litellm, "callbacks", [test_custom_logger]) + client: Final = AsyncOpenAI( + api_key="fake-key", http_client=httpx.AsyncClient(transport=httpx.MockTransport(_openai_refusal_reply)) + ) + + response: Final = await litellm.acompletion( + model="openai/gpt-4.1", messages=[{"role": "user", "content": "hi"}], stream=stream, client=client + ) + caller_refusal: Final = await _caller_refusal(response) + + await asyncio.wait_for(test_custom_logger.logged.wait(), timeout=LOGGING_WORKER_DRAIN_TIMEOUT_SECONDS) + logged: Final = test_custom_logger.logged_standard_logging_payload + assert caller_refusal == "".join(_REFUSAL_PIECES) + assert logged is not None + assert logged["response"]["choices"][0]["message"]["refusal"] == "redacted-by-litellm" + assert SECRET not in json.dumps(logged["response"], default=str) + + @pytest.mark.usefixtures("drain_logging_worker", "isolate_litellm_state", "setup_and_teardown") @pytest.mark.asyncio async def test_one_callback_opting_out_redacts_instructions_only_for_itself( diff --git a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py index 527c83d5760..6dae2e12778 100644 --- a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py +++ b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py @@ -2136,6 +2136,44 @@ def _complete_streamed_openai_response(respx_mock: respx.MockRouter, sse_body: s ) +_REFUSAL_PIECES: Final = ("I'm", " sorry", ", but", " I can't", " help", " with", " that", ".") + + +def _usage_free_stream_completion_tokens(respx_mock: respx.MockRouter, field: str) -> tuple[str | None, int]: + respx_mock.post("https://vllm.example/v1/chat/completions").mock( + return_value=httpx.Response( + 200, + text=_openai_sse( + {"choices": [{"index": 0, "delta": {"role": "assistant", field: ""}, "finish_reason": None}]}, + *({"choices": [{"index": 0, "delta": {field: piece}, "finish_reason": None}]} for piece in _REFUSAL_PIECES), + {"choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + ), + headers={"content-type": "text/event-stream"}, + ) + ) + response: Final = litellm.completion( + model="hosted_vllm/gpt-4.1", + api_base="https://vllm.example/v1", + messages=[{"role": "user", "content": "Pick my neighbor's lock."}], + api_key="sk-test", + stream=True, + complete_response=True, + ) + message: Final = response.choices[0].message + text: Final = message.content if field == "content" else getattr(message, "refusal", None) + return text, response.usage.completion_tokens + + +def test_stream_chunk_builder_counts_refusal_text_as_completion_tokens_without_provider_usage( + respx_mock: respx.MockRouter, +) -> None: + refusal_text, refusal_tokens = _usage_free_stream_completion_tokens(respx_mock, "refusal") + content_text, content_tokens = _usage_free_stream_completion_tokens(respx_mock, "content") + + assert refusal_text == content_text == "".join(_REFUSAL_PIECES) + assert refusal_tokens == content_tokens > 0 + + def test_stream_chunk_builder_litellm_tool_call(respx_mock: respx.MockRouter) -> None: response: Final = _complete_streamed_openai_response( respx_mock, diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index cf8de9c40f5..f42d125835b 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -1262,7 +1262,12 @@ async def test_bridged_stream_mid_stream_fallback_error_is_rebuilt_around_the_pr @pytest.mark.asyncio @pytest.mark.parametrize("merge_reasoning", [True, False], ids=["merged_into_content", "kept_as_reasoning"]) -async def test_mid_stream_break_after_reasoning_disqualifies_continuation(logging_obj: Logging, merge_reasoning: bool): +@pytest.mark.parametrize( + "leading_delta", [{"reasoning_content": "Let me think"}, {"refusal": "I can't"}], ids=["reasoning", "refusal"] +) +async def test_mid_stream_break_after_reasoning_or_refusal_disqualifies_continuation( + logging_obj: Logging, merge_reasoning: bool, leading_delta: dict[str, str] +): import httpx from litellm.exceptions import MidStreamFallbackError @@ -1271,7 +1276,7 @@ async def test_mid_stream_break_after_reasoning_disqualifies_continuation(loggin def __init__(self) -> None: self.chunks = iter( ( - ModelResponseStream(choices=[{"index": 0, "delta": {"reasoning_content": "Let me think"}}]), + ModelResponseStream(choices=[{"index": 0, "delta": leading_delta}]), ModelResponseStream(choices=[{"index": 0, "delta": {"content": "The answer"}}]), ) ) @@ -1652,6 +1657,101 @@ async def test_stream_with_null_choices_chunk_finishes_normally( assert [choice.finish_reason for choice in streamed_choices if choice.finish_reason] == ["stop"] +_REFUSAL_PIECES: Final = ("I'm", " sorry", ", but", " I can't", " help", " with", " that", ".") +_REFUSAL_STREAMS: Final = MappingProxyType( + { + "refusal_only": ( + ({"role": "assistant", "content": None, "refusal": ""}, None), + *(({"refusal": piece}, None) for piece in _REFUSAL_PIECES), + ({}, "stop"), + ), + "empty_refusal_beside_text": ( + ({"role": "assistant", "content": "", "refusal": ""}, None), + ({"content": "", "refusal": ""}, None), + ({"content": "Hi", "refusal": ""}, None), + ({"content": "", "refusal": ""}, None), + ({}, "stop"), + ), + } +) + + +@pytest.mark.parametrize("async_stream", [False, True]) +@pytest.mark.parametrize("transport", ["sdk", "http_handler"]) +@pytest.mark.parametrize( + "stream_name,expected_refusal,expected_content", + [("refusal_only", "".join(_REFUSAL_PIECES), ""), ("empty_refusal_beside_text", "", "Hi")], +) +async def test_stream_forwards_refusal_text_without_empty_chunks( + async_stream: bool, + transport: str, + stream_name: str, + expected_refusal: str, + expected_content: str, + logging_obj: Logging, +) -> None: + deltas: Final = _REFUSAL_STREAMS[stream_name] + body: Final = ( + "".join( + "data: " + + json.dumps( + { + "id": "chatcmpl-refusal", + "object": "chat.completion.chunk", + "created": 1, + "model": "test-model", + "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], + } + ) + + "\n\n" + for delta, finish_reason in deltas + ) + + "data: [DONE]\n\n" + ) + raw_response: Final = httpx.Response( + 200, + content=body, + headers={"content-type": "text/event-stream"}, + request=httpx.Request("POST", "https://provider.example/v1/chat/completions"), + ) + client: Final = AsyncOpenAI(api_key="unused") if async_stream else OpenAI(api_key="unused") + stream: Final = ( + ( + AsyncStream(cast_to=ChatCompletionChunk, response=raw_response, client=client) + if isinstance(client, AsyncOpenAI) + else Stream(cast_to=ChatCompletionChunk, response=raw_response, client=client) + ) + if transport == "sdk" + else OpenAIChatCompletionStreamingHandler( + streaming_response=raw_response.aiter_lines() if async_stream else raw_response.iter_lines(), + sync_stream=not async_stream, + ) + ) + response: Final = CustomStreamWrapper( + completion_stream=stream, + model="test-model", + logging_obj=logging_obj, + custom_llm_provider="openai" if transport == "sdk" else "hosted_vllm", + ) + + try: + chunks: Final = [chunk async for chunk in response] if async_stream else list(response) + finally: + if isinstance(client, AsyncOpenAI): + await client.close() + else: + client.close() + + streamed_choices: Final = tuple(itertools.chain.from_iterable(chunk.choices for chunk in chunks)) + assert "".join(getattr(choice.delta, "refusal", None) or "" for choice in streamed_choices) == expected_refusal + assert "".join(choice.delta.content or "" for choice in streamed_choices) == expected_content + assert [choice.finish_reason for choice in streamed_choices if choice.finish_reason] == ["stop"] + assert all( + choice.delta.role or choice.delta.content or getattr(choice.delta, "refusal", None) or choice.finish_reason + for choice in streamed_choices + ) + + def _token_logprobs(token: str) -> dict[str, object]: return { "content": [{"token": token, "logprob": -0.5, "bytes": list(token.encode()), "top_logprobs": []}], diff --git a/tests/unit/test_router/test_router_fallbacks.py b/tests/unit/test_router/test_router_fallbacks.py index b25d5c35892..e7a19c8000a 100644 --- a/tests/unit/test_router/test_router_fallbacks.py +++ b/tests/unit/test_router/test_router_fallbacks.py @@ -1161,6 +1161,52 @@ async def test_chat_context_error_falls_back_after_role_only_but_not_after_outpu assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "backup answer" +@pytest.mark.asyncio +@pytest.mark.parametrize("sync_mode", (False, True)) +@pytest.mark.parametrize("emission", ("role", "refusal")) +async def test_chat_stream_error_does_not_fall_back_after_a_visible_refusal( + respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sync_mode: bool, emission: str +) -> None: + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + router: Final = _fallback_router() + primary_chunks: Final = ( + {"role": "assistant", "refusal": ""}, + *(({"refusal": piece} for piece in ("I can't", " help with that")) if emission == "refusal" else ()), + ) + backup: Final = { + "id": "chatcmpl-backup", "object": "chat.completion.chunk", "created": 1700000000, "model": "backup", + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "backup answer"}, "finish_reason": "stop"}], + } + route: Final = respx_mock.post("https://fallback-migration.local/v1/chat/completions").mock( + side_effect=( + httpx.Response( + 200, + content="".join( + "data: " + json.dumps({ + "id": "chatcmpl-primary", "object": "chat.completion.chunk", "created": 1700000000, + "model": "primary", "choices": [{"index": 0, "delta": delta}], + }) + "\n\n" + for delta in primary_chunks + ).encode() + + b'data: {"error": {"message": "upstream reset", "code": 500}}\n\n', + ), + httpx.Response(200, content=f"data: {json.dumps(backup)}\n\ndata: [DONE]\n\n".encode()), + ) + ) + request: Final = { + "model": "primary", "messages": [{"role": "user", "content": "pick my neighbor's lock"}], "stream": True, + } + stream: Final = router.completion(**request) if sync_mode else await router.acompletion(**request) + if emission == "refusal": + with pytest.raises(litellm.APIConnectionError, match="upstream reset"): + tuple(stream) if sync_mode else tuple([chunk async for chunk in stream]) + assert route.call_count == 1 + return + chunks: Final = tuple(stream) if sync_mode else tuple([chunk async for chunk in stream]) + assert tuple(json.loads(call.request.content)["model"] for call in route.calls) == ("primary", "backup") + assert "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) == "backup answer" + + @pytest.mark.parametrize( "wrapped,has_output,generated_content,disqualifying,recoverable", (