diff --git a/litellm/__init__.py b/litellm/__init__.py index b6428b51bfb..ad978d37e2f 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -248,6 +248,7 @@ token: Optional[str] = ( max_tokens: int = DEFAULT_MAX_TOKENS # OpenAI Defaults drop_params = drop_params_env_flag(os.environ, verbose_logger) modify_params = bool(os.getenv("LITELLM_MODIFY_PARAMS", False)) +strict_stream_completion: bool = False bedrock_neutralize_orphaned_tool_blocks: bool = True use_chat_completions_url_for_anthropic_messages: bool = bool( os.getenv("LITELLM_USE_CHAT_COMPLETIONS_URL_FOR_ANTHROPIC_MESSAGES", False) @@ -1410,6 +1411,7 @@ from .exceptions import ( APIResponseValidationError, UnprocessableEntityError, InternalServerError, + IncompleteStreamError, JSONSchemaValidationError, LITELLM_EXCEPTION_TYPES, MockException, diff --git a/litellm/exceptions.py b/litellm/exceptions.py index 3bae8a95ef6..84e51d55e6a 100644 --- a/litellm/exceptions.py +++ b/litellm/exceptions.py @@ -888,6 +888,37 @@ class APIConnectionError(openai.APIConnectionError): return _message +class IncompleteStreamError(APIError): + """Raised when a stream ends without a terminal finish_reason or [DONE]. + + Only raised under ``strict_stream_completion``. The default remains the + synthesized ``finish_reason="stop"``, because several backends close the + connection at EOF instead of sending a terminator and would otherwise start + failing. + """ + + def __init__( + self, + message: str, + llm_provider: str, + model: str, + request: httpx.Request | None = None, + litellm_debug_info: str | None = None, + max_retries: int | None = None, + num_retries: int | None = None, + ) -> None: + super().__init__( + status_code=500, + message=message, + llm_provider=llm_provider, + model=model, + request=request or httpx.Request(method="POST", url="https://api.openai.com/v1"), + litellm_debug_info=litellm_debug_info, + max_retries=max_retries, + num_retries=num_retries, + ) + + # raised if an invalid request (not get, delete, put, post) is made class APIResponseValidationError(openai.APIResponseValidationError): def __init__( diff --git a/litellm/litellm_core_utils/get_litellm_params.py b/litellm/litellm_core_utils/get_litellm_params.py index 5adb9a80f9c..71a034066cd 100644 --- a/litellm/litellm_core_utils/get_litellm_params.py +++ b/litellm/litellm_core_utils/get_litellm_params.py @@ -176,6 +176,7 @@ def get_litellm_params( async_call: bool | None = None, ssl_verify: bool | None = None, merge_reasoning_content_in_choices: bool | None = None, + strict_stream_completion: bool | None = None, use_litellm_proxy: bool | None = None, api_version: str | None = None, max_retries: int | None = None, @@ -239,6 +240,7 @@ def get_litellm_params( "async_call": async_call, "ssl_verify": ssl_verify, "merge_reasoning_content_in_choices": merge_reasoning_content_in_choices, + "strict_stream_completion": strict_stream_completion, "api_version": api_version, "max_retries": max_retries, "use_litellm_proxy": use_litellm_proxy, diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 9d33f86d841..2013033041a 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -230,13 +230,21 @@ class CustomStreamWrapper: dict(**self.logging_obj.model_call_details.get("litellm_params", {})) ) self.merge_reasoning_content_in_choices: bool = litellm_params.merge_reasoning_content_in_choices or False + # GenericLiteLLMParams allows extra fields, so the flag arrives without being + # declared on it. The annotation is what keeps this off the unknown-argument + # budget; declaring the field would change the proxy OpenAPI spec. + request_strict: Final[bool | None] = getattr(litellm_params, "strict_stream_completion", None) + self.strict_stream_completion: bool = ( + bool(litellm.strict_stream_completion) if request_strict is None else bool(request_strict) + ) self.sent_first_thinking_block = False self.sent_last_thinking_block = False self.thinking_content = "" self.system_fingerprint: str | None = None self._provider_response_model: str | None = None - self.received_finish_reason: str | None = None + self.stream_reported_finished: bool = False + self.received_finish_reason = None self.intermittent_finish_reason: str | None = None # finish reasons that show up mid-stream self.special_tokens = [ "<|assistant|>", @@ -1712,11 +1720,35 @@ class CustomStreamWrapper: if self.logging_obj._is_sync_litellm_request(litellm_params): self.logging_obj.success_handler(processed_chunk, None, None, cache_hit) + @property + def received_finish_reason(self) -> str | None: + return self._received_finish_reason + + @received_finish_reason.setter + def received_finish_reason(self, value: str | None) -> None: + # A provider assigning this at all is the provider saying the stream + # ended, even when the reason it gives is empty. Recording that here + # rather than beside each of the dozen assignments keeps the two from + # drifting apart. + self._received_finish_reason = value + if value is not None: + self.stream_reported_finished = True + def finish_reason_handler(self): model_response: Final = self.model_response_creator() _finish_reason: Final = self.received_finish_reason or self.intermittent_finish_reason if _finish_reason is not None: model_response.choices[0].finish_reason = _finish_reason + elif self.strict_stream_completion and not self.stream_reported_finished: + raise litellm.exceptions.IncompleteStreamError( + message=( + "Stream ended before a terminal finish_reason or [DONE] delimiter was received. " + "The response is incomplete. Set strict_stream_completion=False to accept it as " + "finished instead." + ), + llm_provider=self.custom_llm_provider or "", + model=self.model or "", + ) else: model_response.choices[0].finish_reason = "stop" diff --git a/litellm/main.py b/litellm/main.py index a818213b861..cfbc7c5c2ac 100644 --- a/litellm/main.py +++ b/litellm/main.py @@ -5658,6 +5658,9 @@ def completion( messages = function_call_prompt(messages=messages, functions=functions_unsupported_model) # For logging - save the values of the litellm-specific params passed in + # kwargs is untyped, so reading the flag straight into the call makes it an + # Any argument; the annotation keeps it off the unknown-argument budget. + strict_stream_completion: bool | None = kwargs.get("strict_stream_completion", None) requested_litellm_params: Final = get_litellm_params( acompletion=acompletion, api_key=api_key, @@ -5695,6 +5698,7 @@ def completion( prompt_variables=prompt_variables, ssl_verify=ssl_verify, merge_reasoning_content_in_choices=kwargs.get("merge_reasoning_content_in_choices", None), + strict_stream_completion=strict_stream_completion, use_litellm_proxy=kwargs.get("use_litellm_proxy", False), api_version=api_version, azure_ad_token=kwargs.get("azure_ad_token"), diff --git a/litellm/types/litellm_params.py b/litellm/types/litellm_params.py index 51f6671e9d6..84ae2f9d89e 100644 --- a/litellm/types/litellm_params.py +++ b/litellm/types/litellm_params.py @@ -239,6 +239,7 @@ class ResponseOptions: merge_reasoning_content_in_choices: bool | None = None enable_json_schema_validation: bool | None = None complete_response: bool | None = None + strict_stream_completion: bool | None = None keepalive_seconds: float | None = None allow_client_keepalive_override: bool | None = None diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index f88e082d577..76ca0bd0f56 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -5143,3 +5143,108 @@ async def test_no_synthetic_finish_reason_logged_when_provider_sent_none(sync_mo assert response.received_finish_reason is None assert all(not (c.choices and c.choices[0].finish_reason) for c in response.chunks) + + +def _strict_wrapper(strict=None, received_finish_reason=None): + logging_obj = MagicMock() + logging_obj.model_call_details = { + "litellm_params": {} if strict is None else {"strict_stream_completion": strict} + } + wrapper = CustomStreamWrapper( + completion_stream=None, + model="gpt-4o", + logging_obj=logging_obj, + custom_llm_provider="openai", + ) + wrapper.received_finish_reason = received_finish_reason + return wrapper + + +def test_truncated_stream_raises_under_strict_stream_completion(): + wrapper = _strict_wrapper(strict=True) + + with pytest.raises(litellm.exceptions.IncompleteStreamError) as exc_info: + wrapper.finish_reason_handler() + + assert exc_info.value.model == "gpt-4o" + assert exc_info.value.llm_provider == "openai" + assert "incomplete" in str(exc_info.value).lower() + + +def test_truncated_stream_still_reports_stop_by_default(): + wrapper = _strict_wrapper() + + assert wrapper.strict_stream_completion is False + assert wrapper.finish_reason_handler().choices[0].finish_reason == "stop" + + +def test_a_finished_stream_is_unaffected_by_strict_stream_completion(): + wrapper = _strict_wrapper(strict=True, received_finish_reason="length") + + assert wrapper.finish_reason_handler().choices[0].finish_reason == "length" + + +def test_strict_stream_completion_reads_an_intermittent_finish_reason(): + wrapper = _strict_wrapper(strict=True) + wrapper.intermittent_finish_reason = "stop" + + assert wrapper.finish_reason_handler().choices[0].finish_reason == "stop" + + +def test_strict_stream_completion_falls_back_to_the_module_setting(monkeypatch): + monkeypatch.setattr(litellm, "strict_stream_completion", True) + assert _strict_wrapper().strict_stream_completion is True + + with pytest.raises(litellm.exceptions.IncompleteStreamError): + _strict_wrapper().finish_reason_handler() + + +def test_the_request_parameter_overrides_the_module_setting(monkeypatch): + monkeypatch.setattr(litellm, "strict_stream_completion", True) + + wrapper = _strict_wrapper(strict=False) + + assert wrapper.strict_stream_completion is False + assert wrapper.finish_reason_handler().choices[0].finish_reason == "stop" + + +def test_incomplete_stream_error_is_retryable_shaped(): + """A 500 is what makes a caller's existing retry policy pick this up""" + error = litellm.exceptions.IncompleteStreamError("cut", llm_provider="openai", model="gpt-4o") + + assert error.status_code == 500 + assert isinstance(error, litellm.exceptions.APIError) + + +def test_a_clean_terminator_with_no_finish_reason_is_not_treated_as_truncation(): + """A provider can report the stream finished while leaving finish_reason empty""" + wrapper = _strict_wrapper(strict=True) + wrapper.stream_reported_finished = True + + assert wrapper.finish_reason_handler().choices[0].finish_reason == "stop" + + +def test_incomplete_stream_error_is_reachable_from_the_package_root(): + """A caller writes `except litellm.IncompleteStreamError`, not the submodule path""" + assert litellm.IncompleteStreamError is litellm.exceptions.IncompleteStreamError + + +def test_an_ordinary_content_chunk_does_not_mark_the_stream_finished(): + """The nlp_cloud branch set the flag for every parsed chunk, not only a terminal one""" + wrapper = _strict_wrapper(strict=True) + wrapper.custom_llm_provider = "nlp_cloud" + + assert wrapper.stream_reported_finished is False + wrapper.received_finish_reason = None + assert wrapper.stream_reported_finished is False + + with pytest.raises(litellm.exceptions.IncompleteStreamError): + wrapper.finish_reason_handler() + + +def test_an_empty_finish_reason_still_marks_the_stream_finished(): + wrapper = _strict_wrapper(strict=True) + wrapper.received_finish_reason = "" + + assert wrapper.stream_reported_finished is True + assert wrapper.finish_reason_handler().choices[0].finish_reason == "stop" diff --git a/tests/unit/types/test_litellm_params.py b/tests/unit/types/test_litellm_params.py index e3bbae39468..b5a8ba4aebf 100644 --- a/tests/unit/types/test_litellm_params.py +++ b/tests/unit/types/test_litellm_params.py @@ -185,6 +185,7 @@ OPTION_NAMES: Final = ( "merge_reasoning_content_in_choices", "enable_json_schema_validation", "complete_response", + "strict_stream_completion", "stream_chunk_size", "keepalive_seconds", "allow_client_keepalive_override",