scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-14 09:03:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 214

fix(proxy): name the deployment in stream error alerts on every provider and route

Browse source 
The llm_exceptions alert for a cut stream on Anthropic chat, /v1/completions,
native /v1/responses and native /v1/messages had no API Base, model_group or
deployment line. The shared exception mapper now attaches its debug info at its
single exit, mid-stream fallback errors inherit it from the error they wrap,
and the native relays add it to the same error object they re-raise.
This commit is contained in:
mateo-berri 2026-10-29 04:32:37 -07:00
parent 5eb1042bbf
commit fa152013a8
12 changed files with 340 additions and 37 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
litellm/exceptions.py
View file

@ -1236,7 +1236,10 @@ class MidStreamFallbackError(ServiceUnavailableError):
self.model = model
self.llm_provider = llm_provider
self.original_exception = original_exception
self.litellm_debug_info = litellm_debug_info
inherited_debug_info: Final[object] = getattr(original_exception, "litellm_debug_info", None)
self.litellm_debug_info = litellm_debug_info or (
inherited_debug_info if isinstance(inherited_debug_info, str) else None
)
self.max_retries = max_retries
self.num_retries = num_retries
self.generated_content = generated_content

37
litellm/litellm_core_utils/exception_mapping_utils.py
View file

@ -2443,6 +2443,39 @@ def with_deployment_api_base(litellm_params: object) -> Mapping[str, object]:
return MappingProxyType({**params, "api_base": deployment_api_base})
_DEBUG_INFO_ATTRIBUTE: Final = "litellm_debug_info"
def _attach_debug_info(exception: Exception, debug_info: str) -> None:
if debug_info and getattr(exception, _DEBUG_INFO_ATTRIBUTE, None) is None:
setattr(exception, _DEBUG_INFO_ATTRIBUTE, debug_info)
def mapped_stream_exception(
original_exception: Exception, model: str, custom_llm_provider: str, litellm_params: object
) -> Exception:
try:
return exception_type(
model=model,
custom_llm_provider=custom_llm_provider,
original_exception=original_exception,
completion_kwargs={},
extra_kwargs=with_deployment_api_base(litellm_params),
)
except Exception as mapped_exception:
return mapped_exception
def with_deployment_debug_info(
error: Exception, model: str, custom_llm_provider: str, litellm_params: object
) -> Exception:
debug_info: Final[object] = getattr(
mapped_stream_exception(error, model, custom_llm_provider, litellm_params), _DEBUG_INFO_ATTRIBUTE, None
)
_attach_debug_info(error, debug_info if isinstance(debug_info, str) else "")
return error
def _api_base_without_credentials(api_base: str) -> str:
scheme, separator, _ = api_base.partition("://")
scheme_prefix: Final = scheme + separator if separator and _URL_SCHEME.fullmatch(scheme) else ""
@ -2492,11 +2525,11 @@ def exception_type(
print() # noqa: T201
litellm_response_headers: Final = _get_response_headers(original_exception=original_exception)
extra_information = ""
try:
error_str = (
redact_secret_string(str(original_exception)) if _ENABLE_SECRET_REDACTION else str(original_exception)
)
extra_information = ""
if model or custom_llm_provider:
if hasattr(original_exception, "message"):
error_str = (
@ -2837,11 +2870,13 @@ def exception_type(
# don't let an error with mapping interrupt the user from receiving an error from the llm api calls
if exception_mapping_worked:
setattr(e, "litellm_response_headers", litellm_response_headers)
_attach_debug_info(e, extra_information)
raise e
else:
for error_type in litellm.LITELLM_EXCEPTION_TYPES:
if isinstance(e, error_type):
setattr(e, "litellm_response_headers", litellm_response_headers)
_attach_debug_info(e, extra_information)
raise e # it's already mapped
raised_exc: Final = APIConnectionError(
message=f"{original_exception}\n{redact_string(traceback.format_exc())}",

26
litellm/llms/anthropic/messages/native/streaming_iterator.py
View file

@ -15,6 +15,7 @@ from litellm.constants import (
ANTHROPIC_MESSAGES_STREAM_RELAY_QUEUE_MAXSIZE,
)
from litellm.litellm_core_utils.core_helpers import process_response_headers
from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_debug_info
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.llms.anthropic.common_utils import ANTHROPIC_ERROR_STATUS_CODE_MAP
@ -562,6 +563,7 @@ class BaseAnthropicMessagesStreamingIterator:
start_time=self.start_time,
passthrough_success_handler_obj=GLOBAL_PASS_THROUGH_SUCCESS_HANDLER_OBJ,
url_route="/v1/messages",
on_upstream_error=self._with_deployment_debug_info,
)
def _convert_chunk_to_sse_format(self, chunk: dict | Any) -> bytes:
@ -609,8 +611,9 @@ class BaseAnthropicMessagesStreamingIterator:
An upstream failure (Bedrock read / decode / chunk-conversion error)
that happens while the client is still connected is forwarded through
the queue and re-raised here, so the original provider exception (and
its status) reaches the proxy's failure handling unchanged rather than
being masked by a generic incomplete-stream event.
its status) reaches the proxy's failure handling with the deployment's
debug info attached, rather than being masked by a generic
incomplete-stream event.
This method provides the common logic for both Anthropic and Bedrock implementations.
"""
@ -797,18 +800,29 @@ class BaseAnthropicMessagesStreamingIterator:
"""
from litellm.proxy.pass_through_endpoints.streaming_handler import PassThroughStreamingHandler
described: Final = self._with_deployment_debug_info(exc)
await PassThroughStreamingHandler.schedule_stream_failure_logging(
litellm_logging_obj=self.litellm_logging_obj,
endpoint_type=EndpointType.ANTHROPIC,
request_body=self.request_body,
raw_bytes=collected_chunks,
exception=exc,
exception=described,
)
if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, exc):
if not client_detached.is_set() and await self._enqueue_for_client(queue, client_detached, described):
await client_detached.wait()
if not _exception_left_unconsumed(queue, exc):
if not _exception_left_unconsumed(queue, described):
return
await self._fire_detached_failure_hook(exc)
await self._fire_detached_failure_hook(described)
def _with_deployment_debug_info(self, exc: Exception) -> Exception:
details: Final = self.litellm_logging_obj.model_call_details
provider: Final = details.get("custom_llm_provider")
return with_deployment_debug_info(
exc,
self.litellm_logging_obj.model,
provider if isinstance(provider, str) and provider else "anthropic",
details.get("litellm_params"),
)
async def _fire_detached_failure_hook(self, exc: Exception) -> None:
from litellm._logging import verbose_proxy_logger

15
litellm/main.py
View file

@ -79,6 +79,7 @@ from litellm.litellm_core_utils.chat_completion_agentic_loop import (
)
from litellm.litellm_core_utils.completion_timeout import CompletionTimeout, StreamKind
from litellm.litellm_core_utils.dd_tracing import tracer
from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception
from litellm.litellm_core_utils.get_litellm_params import (
AWS_CREDENTIAL_KWARGS_KEYS,
OAUTH_TOKEN_EXCHANGE_KWARGS_KEYS,
@ -817,7 +818,7 @@ def _text_completion_response_from_cached_dict(cached_response_dict: Mapping[str
return TextCompletionResponse(stream=bool(cached_response_dict.get("stream", False)), **fields)
async def _async_streaming(response, model, custom_llm_provider, args):
async def _async_streaming(response, model, custom_llm_provider, litellm_params: Mapping[str, object]):
try:
print_verbose(f"received response in _async_streaming: {response}")
if asyncio.iscoroutine(response):
@ -826,11 +827,11 @@ async def _async_streaming(response, model, custom_llm_provider, args):
print_verbose(f"line in async streaming: {line}")
yield line
except Exception as e:
custom_llm_provider = custom_llm_provider or "openai"
raise exception_type(
model=model,
custom_llm_provider=custom_llm_provider,
original_exception=e,
raise mapped_stream_exception(
e,
model if isinstance(model, str) else "",
custom_llm_provider if isinstance(custom_llm_provider, str) and custom_llm_provider else "openai",
litellm_params,
)
@ -7804,7 +7805,7 @@ async def atext_completion(*args, **kwargs) -> TextCompletionResponse | TextComp
response=response,
model=model,
custom_llm_provider=custom_llm_provider,
args=args,
litellm_params=_REQUEST_KWARGS_ADAPTER.validate_python(kwargs),
),
model=model,
custom_llm_provider=custom_llm_provider,

2
litellm/proxy/pass_through_endpoints/streaming_handler.py
View file

@ -166,6 +166,7 @@ class PassThroughStreamingHandler:
passthrough_success_handler_obj: PassThroughEndpointLogging,
url_route: str,
route_streaming_logging: RouteStreamingLogging | None = None,
on_upstream_error: Callable[[Exception], object] = lambda _: None,
):
resolved_route_streaming_logging: Final[RouteStreamingLogging] = (
route_streaming_logging or PassThroughStreamingHandler.route_streaming_logging_to_handler
@ -265,6 +266,7 @@ class PassThroughStreamingHandler:
litellm_logging_obj._deferred_stream_complete_args = (_build_logging_coroutine(),)
except Exception as e:
verbose_proxy_logger.error("Error in chunk_processor: %s", e)
on_upstream_error(e)
if response.status_code < 400:
logging_scheduled = True
await PassThroughStreamingHandler.schedule_stream_failure_logging(

33
litellm/responses/streaming_iterator.py
View file

@ -44,7 +44,7 @@ from litellm.litellm_core_utils.core_helpers import (
provider_reported_usage_cost,
set_response_cost_in_hidden_params,
)
from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base
from litellm.litellm_core_utils.exception_mapping_utils import mapped_stream_exception, with_deployment_debug_info
from litellm.litellm_core_utils.internal_call_metadata import (
is_background_response,
requests_background,
@ -318,27 +318,21 @@ def map_stream_error_to_exception(
return _mapped_exception(provider_exception, model, custom_llm_provider, logging_obj)
def _logged_litellm_params(logging_obj: LiteLLMLoggingObj | None) -> object:
return (
None
if logging_obj is None
else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {})
)
def _mapped_exception(
original_exception: Exception,
model: str,
custom_llm_provider: str,
logging_obj: LiteLLMLoggingObj | None,
) -> Exception:
litellm_params: Final = (
None
if logging_obj is None
else _typed_gets_litellm_params(logging_obj.model_call_details.get)("litellm_params", {})
)
try:
return litellm.exception_type(
model=model,
custom_llm_provider=custom_llm_provider,
original_exception=original_exception,
completion_kwargs={},
extra_kwargs=with_deployment_api_base(litellm_params),
)
except Exception as mapped_exception:
return mapped_exception
return mapped_stream_exception(original_exception, model, custom_llm_provider, _logged_litellm_params(logging_obj))
def _event_field(event: object, name: str) -> object:
@ -1175,9 +1169,12 @@ class BaseResponsesAPIStreamingIterator:
def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn:
self._handle_failure(error)
described: Final = with_deployment_debug_info(
error, self.model or "", self.custom_llm_provider or "", _logged_litellm_params(self.logging_obj)
)
if self._output_started:
raise error
raise self._fallback_error(error) from error
raise described
raise self._fallback_error(described) from described
async def call_post_streaming_hooks_for_testing(

5
litellm/utils.py
View file

@ -6008,9 +6008,8 @@ def get_model_region(litellm_params: LiteLLM_Params, mode: str | None) -> str |
def get_first_chars_messages(kwargs: dict) -> str:
try:
_messages = kwargs.get("messages")
_messages = str(_messages)[:100]
return _messages
messages: Final = kwargs.get("messages")
return "" if messages is None else str(messages)[:100]
except Exception:
return ""

100
tests/integration/observability/test_stream_error_alert_api_base.py
View file

@ -13,7 +13,15 @@ import openai
import pytest
import yaml
from anthropic.types import RawContentBlockDeltaEvent, TextDelta
from integration._support.anthropic_sse import DROP_PAUSE, dropping_reply, stream_reply
from integration._support.anthropic_sse import (
CONTENT_BLOCK_START,
DROP_PAUSE,
dropping_reply,
error_body,
message_start,
stream_reply,
text_delta,
)
from integration._support.client import (
JSON_OBJECT,
Gateway,
@ -30,6 +38,7 @@ from integration._support.responses_ws_peer import PROVIDER_MODEL, Script, event
from integration._support.tls import write_self_signed_cert
from integration._support.vertex import service_account_json
from integration._support.wire import Reply, Request, Wire, wire_server
from openai.types import Completion
from openai.types.chat import ChatCompletionChunk
from openai.types.responses import ResponseFailedEvent, ResponseTextDeltaEvent
from pydantic import JsonValue
@ -39,6 +48,7 @@ pytestmark: Final = [pytest.mark.timeout(2 * graceful_stop_seconds() + 120), pyt
AZURE_KEY: Final = "synthetic-azure-key"
API_VERSION: Final = "2025-04-01-preview"
AZURE_CHAT_BACKEND: Final = "azure/gpt-4o-mini"
ANTHROPIC_BACKEND: Final = "anthropic/claude-opus-5-5"
PARTIAL_TEXT: Final = "partial answer"
ALERT_SECONDS: Final = 40
NO_RETRY_NO_CACHE: Final[Mapping[str, JsonValue]] = MappingProxyType({"num_retries": 0, "cache": {"no-cache": True}})
@ -234,6 +244,10 @@ def responses_text(item: object) -> str:
return item.delta if isinstance(item, ResponseTextDeltaEvent) else ""
def completion_text(item: object) -> str:
return "".join(choice.text for choice in item.choices) if isinstance(item, Completion) else ""
def failure_of(item: object) -> str:
if isinstance(item, Exception):
return str(item)
@ -296,6 +310,19 @@ def responses_outcome(gateway: Gateway, key: str, model: str) -> Outcome:
return outcome(items, responses_text)
def completions_outcome(gateway: Gateway, key: str, model: str) -> Outcome:
with openai_client(gateway, key) as client:
items: Final = tuple(
items_or_error(
lambda: client.completions.create(
model=model, prompt="stream an answer", stream=True, extra_body=dict(NO_RETRY_NO_CACHE)
),
openai.APIError,
)
)
return outcome(items, completion_text)
ENTRYPOINTS: Final[Mapping[str, Callable[[Gateway, str, str], Outcome]]] = MappingProxyType(
{"chat": chat_outcome, "messages": messages_outcome, "responses-bridge": responses_outcome}
)
@ -480,3 +507,74 @@ def test_fallback_mid_stream_failure_alert_names_the_fallback_deployment(alertin
alert: Final = alert_naming(alerting.alerts, marker)
assert_names_deployment(alert, backup_wire.url, backup, AZURE_CHAT_BACKEND, caller)
assert primary_wire.url not in alert, alert
PING_NEVER_SENT: Final = b"event: ping\ndata: {}\n\n"
def anthropic_cut_after_text(marker: str) -> Reply:
return dropping_reply(
(
message_start(f"msg_{marker}", ANTHROPIC_BACKEND),
CONTENT_BLOCK_START,
text_delta(PARTIAL_TEXT),
PING_NEVER_SENT,
),
abort_after=3,
)
def responses_cut_after_text(marker: str) -> Reply:
identity: Final = f"resp_{marker}"
return dropping_reply(
(frame(created(identity)), frame(delta(identity, PARTIAL_TEXT)), b": never sent\n\n"), abort_after=2
)
@dataclass(frozen=True, slots=True)
class CutPath:
backend: str
reply: Callable[[str], Reply]
call: Callable[[Gateway, str, str], Outcome]
CUT_PATHS: Final[Mapping[str, CutPath]] = MappingProxyType(
{
"anthropic-chat": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, chat_outcome),
"completions": CutPath(AZURE_CHAT_BACKEND, cut_after_text, completions_outcome),
"native-responses": CutPath(AZURE_CHAT_BACKEND, responses_cut_after_text, responses_outcome),
"native-messages": CutPath(ANTHROPIC_BACKEND, anthropic_cut_after_text, messages_outcome),
}
)
def scripted_model(scenario: Scenario, backend: str, api_base: str) -> str:
if backend == AZURE_CHAT_BACKEND:
return azure_chat(scenario, api_base)
return scenario.model(model=backend, api_base=api_base, api_key="synthetic-anthropic-key", num_retries=0)
@pytest.mark.parametrize("path", list(CUT_PATHS))
def test_stream_cut_alert_names_the_deployment(alerting: AlertingProxy, path: str) -> None:
marker: Final = uuid.uuid4().hex
cut: Final = CUT_PATHS[path]
with wire_server(chat_upstream(cut.reply, marker)) as wire, alerting.gateway.scenario() as scenario:
model: Final = scripted_model(scenario, cut.backend, wire.url)
caller: Final = keyed_caller(scenario)
seen: Final = cut.call(alerting.gateway, caller.key, model)
assert seen.text == PARTIAL_TEXT and seen.failure, seen
assert_names_deployment(alert_naming(alerting.alerts, model), wire.url, model, cut.backend, caller)
def test_anthropic_rejection_before_stream_alert_names_the_deployment(alerting: AlertingProxy) -> None:
marker: Final = uuid.uuid4().hex
def rejected(_: str) -> Reply:
return Reply(status=500, body=error_body(500, provider_message(marker)))
with wire_server(chat_upstream(rejected, marker)) as wire, alerting.gateway.scenario() as scenario:
model: Final = scripted_model(scenario, ANTHROPIC_BACKEND, wire.url)
caller: Final = keyed_caller(scenario)
seen: Final = chat_outcome(alerting.gateway, caller.key, model)
assert provider_message(marker) in seen.failure and not seen.text, seen
assert_names_deployment(alert_naming(alerting.alerts, marker), wire.url, model, ANTHROPIC_BACKEND, caller)

36
tests/unit/litellm_core_utils/test_exception_mapping_utils.py
View file

@ -20,6 +20,7 @@ from litellm.litellm_core_utils.exception_mapping_utils import (
exception_type,
extract_and_raise_litellm_exception,
)
from litellm.llms.anthropic.common_utils import AnthropicError
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.bedrock.common_utils import BedrockError
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
@ -156,6 +157,41 @@ def test_api_base_line_keeps_an_at_sign_in_the_path(api_base: str) -> None:
assert f"API Base: `{api_base}`" in _debug_info_for_api_base(api_base)
@pytest.mark.parametrize(
("provider_error", "mapped_class"),
[
(
httpx.ReadError("peer closed connection", request=httpx.Request("POST", "https://relay.example/v1/messages")),
litellm.APIConnectionError,
),
(RuntimeError("stream decoder failed"), litellm.APIConnectionError),
(AnthropicError(status_code=500, message="[Errno 104] Connection reset by peer"), litellm.InternalServerError),
(AnthropicError(status_code=429, message="rate_limit_error"), litellm.RateLimitError),
],
ids=["unmapped_with_request", "unmapped_without_request", "anthropic_500", "anthropic_429"],
)
def test_mapped_provider_error_carries_the_deployment_debug_info(
provider_error: Exception, mapped_class: type[Exception]
) -> None:
with pytest.raises(mapped_class) as caught:
exception_type(
model="claude-opus-5-5",
original_exception=provider_error,
custom_llm_provider="anthropic",
completion_kwargs={},
extra_kwargs={
"api_base": "https://relay.example",
"metadata": {"model_group": "claude-alias", "deployment": "anthropic/claude-opus-5-5"},
},
)
debug_info: Final = caught.value.litellm_debug_info
assert isinstance(debug_info, str)
assert "API Base: `https://relay.example`" in debug_info
assert "model_group: `claude-alias`" in debug_info
assert "deployment: `anthropic/claude-opus-5-5`" in debug_info
assert "Messages:" not in debug_info
# Test cases for is_error_str_context_window_exceeded
# Tuple format: (error_message, expected_result)
context_window_test_cases = [

61
tests/unit/llms/anthropic/messages/native/test_streaming_iterator.py
View file

@ -645,6 +645,67 @@ async def test_async_sse_wrapper_reraises_upstream_error_to_connected_client():
assert detached_hook.exceptions == [], "the relay re-raised the error, so the proxy failure hook already ran"
_DEPLOYMENT_PARAMS: Final = {
"metadata": {
"api_base": "https://relay.example/anthropic",
"model_group": "claude-alias",
"deployment": "anthropic/claude-opus-5-5",
}
}
def _assert_names_the_deployment(error: Exception) -> None:
debug_info: Final = getattr(error, "litellm_debug_info", None)
assert isinstance(debug_info, str)
assert "API Base: `https://relay.example/anthropic`" in debug_info
assert "model_group: `claude-alias`" in debug_info
assert "deployment: `anthropic/claude-opus-5-5`" in debug_info
@pytest.mark.asyncio
async def test_relayed_upstream_error_reaches_the_client_naming_the_deployment():
async def _cut_stream():
yield {"type": "message_start", "message": {"id": "msg_1", "usage": {"input_tokens": 52, "output_tokens": 1}}}
raise _ProviderStreamError("Response payload is not completed", status_code=529)
iterator = _RecordingLoggingIterator(
litellm_logging_obj=_make_logging_obj("test_upstream_error_names_the_deployment"),
request_body={},
)
iterator.litellm_logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS
with pytest.raises(_ProviderStreamError) as excinfo:
async for _ in iterator.async_sse_wrapper(_cut_stream()):
pass
assert excinfo.value.status_code == 529
_assert_names_the_deployment(excinfo.value)
@pytest.mark.asyncio
async def test_native_anthropic_stream_cut_reaches_the_client_naming_the_deployment() -> None:
logging_obj: Final = _make_logging_obj("native_anthropic_stream_cut")
logging_obj.model_call_details["litellm_params"] = _DEPLOYMENT_PARAMS
cut: Final = httpx.ReadError("Response payload is not completed")
async def _cut_upstream() -> AsyncIterator[bytes]:
yield b'event: message_start\ndata: {"type": "message_start"}\n\n'
raise cut
stream: Final = AnthropicMessagesConfig().get_async_streaming_response_iterator(
model=logging_obj.model,
httpx_response=httpx.Response(200, content=_cut_upstream()),
request_body={"model": logging_obj.model},
litellm_logging_obj=logging_obj,
)
with pytest.raises(httpx.ReadError) as excinfo:
_ = [chunk async for chunk in stream]
assert excinfo.value is cut
_assert_names_the_deployment(cut)
@pytest.mark.asyncio
async def test_async_sse_wrapper_logs_failure_on_upstream_error_after_disconnect():
"""

28
tests/unit/responses/test_streaming_iterator.py
View file

@ -365,6 +365,34 @@ def test_sync_transport_read_timeout_raises_litellm_timeout(prefix):
assert "Request timed out. Error_str: ReadTimeout" in str(exc_info.value)
@pytest.mark.asyncio
@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES)
@pytest.mark.parametrize("error_class", [httpx.ReadError, httpx.RemoteProtocolError])
async def test_transport_cut_error_names_the_deployment(prefix, error_class):
logging_obj: Final = _logging_obj_stub()
logging_obj.model_call_details = {
"litellm_params": {
"metadata": {
"api_base": "https://relay.example/openai/v1",
"model_group": "responses-alias",
"deployment": "openai/gpt-4o-mini",
}
}
}
trailing_error: Final = error_class("peer closed connection without sending complete message body")
iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj, trailing_error=trailing_error)
with pytest.raises((error_class, MidStreamFallbackError)) as exc_info:
async for _ in iterator:
pass
debug_info: Final = exc_info.value.litellm_debug_info
assert isinstance(debug_info, str)
assert "API Base: `https://relay.example/openai/v1`" in debug_info
assert "model_group: `responses-alias`" in debug_info
assert "deployment: `openai/gpt-4o-mini`" in debug_info
@pytest.mark.asyncio
@pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"])
async def test_stream_ending_after_partial_output_without_terminal_event_raises(trailer):

29
tests/unit/text_completions/test_streaming_iterator.py
View file

@ -1,9 +1,11 @@
from collections.abc import AsyncIterator, Iterator
from typing import Final
import httpx
import pytest
from litellm.exceptions import APIConnectionError
from litellm.main import _async_streaming
from litellm.text_completions.streaming_iterator import TextCompletionStreamWrapper
from litellm.types.utils import Choices, Delta, ModelResponseStream, StreamingChoices, Usage
@ -142,3 +144,30 @@ def test_sync_text_completion_stream_maps_upstream_failure_after_content() -> No
next(wrapper)
assert (error.value.model, error.value.llm_provider) == ("test-model", "openai")
_DEPLOYMENT_PARAMS: Final = {
"api_base": "https://relay.example/v1",
"metadata": {"model_group": "instruct-alias", "deployment": "text-completion-openai/gpt-instruct"},
}
@pytest.mark.asyncio
async def test_async_text_completion_upstream_cut_names_the_deployment() -> None:
async def cut_stream() -> AsyncIterator[ModelResponseStream]:
raise httpx.ReadError("Response payload is not completed")
yield
wrapper: Final = TextCompletionStreamWrapper(
completion_stream=_async_streaming(cut_stream(), "test-model", None, _DEPLOYMENT_PARAMS),
model="test-model",
)
with pytest.raises(APIConnectionError, match="OpenAIException") as error:
await anext(wrapper)
debug_info: Final = error.value.litellm_debug_info
assert isinstance(debug_info, str)
assert "API Base: `https://relay.example/v1`" in debug_info
assert "model_group: `instruct-alias`" in debug_info
assert "deployment: `text-completion-openai/gpt-instruct`" in debug_info