Merge 88a7f10ad4 into 8364f88cbb

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,

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__(

litellm/litellm_core_utils/get_litellm_params.py

@@ -179,6 +179,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,
@@ -242,6 +243,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,

litellm/litellm_core_utils/streaming_handler.py

@@ -230,13 +230,18 @@ 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
+        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 +1717,31 @@ 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:
+        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"
 
@@ -1929,7 +1954,16 @@ class CustomStreamWrapper:
                 raise  # Re-raise StopIteration
             else:
                 self.sent_last_chunk = True
-                processed_chunk: Final = self.finish_reason_handler()
+                try:
+                    processed_chunk: Final = self.finish_reason_handler()
+                except litellm.exceptions.IncompleteStreamError as strict_error:
+                    self._record_partial_usage_for_failure()
+                    if self.logging_obj is not None:
+                        threading.Thread(
+                            target=self.logging_obj.failure_handler,
+                            args=(strict_error, traceback.format_exc()),
+                        ).start()
+                    raise
                 # The logged response is built from self.chunks; keep a finish_reason the provider sent on its
                 # last content chunk (stripped there), but never add the synthetic "stop" used when it sent none.
                 if self.received_finish_reason is not None or self.intermittent_finish_reason is not None:
@@ -2197,7 +2231,17 @@ class CustomStreamWrapper:
             raise StopAsyncIteration  # Re-raise StopIteration
         else:
             self.sent_last_chunk = True
-            processed_chunk: Final = self.finish_reason_handler()
+            try:
+                processed_chunk: Final = self.finish_reason_handler()
+            except litellm.exceptions.IncompleteStreamError as strict_error:
+                self._record_partial_usage_for_failure()
+                if self.logging_obj is not None:
+                    asyncio.create_task(
+                        self.logging_obj.dispatch_failure_handlers(
+                            strict_error, traceback.format_exc(), prefer_async_handlers=True
+                        )
+                    )
+                raise
             # The logged response is built from self.chunks; keep a finish_reason the provider sent on its
             # last content chunk (stripped there), but never add the synthetic "stop" used when it sent none.
             if self.received_finish_reason is not None or self.intermittent_finish_reason is not None:

litellm/main.py

@@ -5662,6 +5662,7 @@ def completion(
             messages = function_call_prompt(messages=messages, functions=functions_unsupported_model)
 
         # For logging - save the values of the litellm-specific params passed in
+        strict_stream_completion: bool | None = kwargs.get("strict_stream_completion", None)
         requested_litellm_params: Final = get_litellm_params(
             acompletion=acompletion,
             api_key=api_key,
@@ -5699,6 +5700,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"),

litellm/types/litellm_params.py

@@ -270,6 +270,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
 

tests/unit/litellm_core_utils/test_streaming_handler.py

@@ -5143,3 +5143,145 @@ 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"
+
+
+@pytest.mark.parametrize("sync_mode", [True, False])
+@pytest.mark.asyncio
+async def test_strict_incomplete_stream_records_partial_usage_before_raising(
+    sync_mode: bool, logging_obj: Logging
+):
+    chunks = [
+        ModelResponseStream(
+            id="chatcmpl-1",
+            created=1,
+            model=None,
+            object="chat.completion.chunk",
+            choices=[
+                StreamingChoices(
+                    finish_reason=None, index=0, delta=Delta(content="partial", role="assistant")
+                )
+            ],
+        )
+    ]
+    response = CustomStreamWrapper(
+        completion_stream=ModelResponseListIterator(model_responses=chunks),
+        model="gpt-4o",
+        custom_llm_provider="openai",
+        logging_obj=logging_obj,
+    )
+    response.strict_stream_completion = True
+    recorded = MagicMock()
+    response._record_partial_usage_for_failure = recorded
+
+    with pytest.raises(litellm.exceptions.IncompleteStreamError):
+        if sync_mode:
+            list(response)
+        else:
+            [c async for c in response]
+
+    recorded.assert_called_once()

tests/unit/types/test_litellm_params.py

@@ -206,6 +206,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",