mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
fix(proxy): name the deployment in stream error alerts on chat, completions, responses and messages (#51034)
* fix(proxy): name the deployment in stream error alerts on every provider and route 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. * fix(proxy): keep the prompt out of attached alert debug info and name the deployment on more stream exits * fix(proxy): name the deployment on stream errors that arrive already mapped exception_type returned errors that were already litellm types before it attached the debug info, so a finish_reason=abort chunk or the max streaming duration timeout still sent an alert with no deployment. It now builds the info first and attaches the prompt-free copy on that early return too, and native Responses streams attach it on their generic error exit Also adds tests for the dropped `Messages: None` line and the mapping-crash exit * fix(proxy): read the router's deployment from litellm_metadata on Messages and Responses errors /v1/messages and /v1/responses keep the router's model_group, deployment and api_base in litellm_metadata, so their errors before the stream named only the model in alerts. The debug info now reads litellm_metadata first, falls back to metadata, and takes API Base from it when the call passes none get_first_chars_messages returns an empty string again for input that is not a mapping, as it did before this branch * fix(proxy): merge both metadata buckets when error alerts name the deployment A pre_call guardrail adds litellm_metadata to a chat request while the router keeps model_group and deployment in metadata, and generic routes such as moderations keep the key alias and team in metadata and the deployment in litellm_metadata. Reading only one bucket dropped lines the merge base showed. The debug info now merges both, and litellm_metadata wins when both carry the same key, so a caller's own metadata cannot replace the router's values * fix(proxy): keep a reused error's alert details on the request that raised it An exception instance raised for more than one request, such as an SDK mock_response exception or a handler that raises one shared error, kept the first request's Key Name, model_group and deployment in every later alert The attached details now record the call id that set them. A later call with a different id replaces them, and a layer that re-maps the error without the request keeps them * fix(proxy): drop alert details from an error object raised by more than one request One exception object raised by two concurrent requests (a shared mock_response error, or a handler that raises one error object for every request) let a request's alert show the other request's Key Name and deployment, since the later attach replaced the earlier one. An error object tagged by one call id and attached again under another now drops its details for good, so no alert names the wrong request. One request re-mapping its own error, with or without a call id, keeps the lines it already has --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
quietjules
GitHub
parent
3671d6b380
commit
534a60d28d
13 changed files with 762 additions and 102 deletions
|
|
@ -1289,7 +1289,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
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@ import json
|
|||
import re
|
||||
import traceback
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Protocol, cast
|
||||
|
||||
|
|
@ -2514,15 +2515,113 @@ _HOST_WITH_OPTIONAL_PORT: Final = re.compile(
|
|||
)
|
||||
|
||||
|
||||
def _router_metadata(params: Mapping[str, object]) -> Mapping[str, object]:
|
||||
metadata: Final = as_str_mapping(params.get("metadata")) or _NO_LITELLM_PARAMS
|
||||
litellm_metadata: Final = as_str_mapping(params.get("litellm_metadata")) or _NO_LITELLM_PARAMS
|
||||
return MappingProxyType({**metadata, **litellm_metadata})
|
||||
|
||||
|
||||
def _deployment_api_base(metadata: Mapping[str, object]) -> str | None:
|
||||
api_base: Final = metadata.get("api_base")
|
||||
return api_base if isinstance(api_base, str) and api_base else None
|
||||
|
||||
|
||||
def with_deployment_api_base(litellm_params: object) -> Mapping[str, object]:
|
||||
params: Final = as_str_mapping(litellm_params) or _NO_LITELLM_PARAMS
|
||||
metadata: Final = as_str_mapping(params.get("metadata"))
|
||||
deployment_api_base: Final = None if metadata is None else metadata.get("api_base")
|
||||
if not isinstance(deployment_api_base, str) or not deployment_api_base:
|
||||
deployment_api_base: Final = _deployment_api_base(_router_metadata(params))
|
||||
if deployment_api_base is None:
|
||||
return params
|
||||
return MappingProxyType({**params, "api_base": deployment_api_base})
|
||||
|
||||
|
||||
_DEBUG_INFO_ATTRIBUTE: Final = "litellm_debug_info"
|
||||
_DEBUG_INFO_CALL_ID_ATTRIBUTE: Final = "_litellm_debug_info_call_id"
|
||||
_RAISED_BY_SEVERAL_CALLS: Final = object()
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _RequestDebugInfo:
|
||||
with_prompt: str
|
||||
without_prompt: str
|
||||
call_id: str
|
||||
|
||||
|
||||
_NO_REQUEST_DEBUG_INFO: Final = _RequestDebugInfo(with_prompt="", without_prompt="", call_id="")
|
||||
|
||||
|
||||
def _request_debug_info(model: object, extra_kwargs: object, completion_kwargs: object) -> _RequestDebugInfo:
|
||||
params: Final = as_str_mapping(extra_kwargs) or _NO_LITELLM_PARAMS
|
||||
try:
|
||||
metadata: Final = _router_metadata(params)
|
||||
api_base: Final = litellm.get_api_base(model=model, optional_params=params) or _deployment_api_base(metadata)
|
||||
messages: Final = litellm.get_first_chars_messages(
|
||||
kwargs=as_str_mapping(completion_kwargs) or _NO_LITELLM_PARAMS
|
||||
)
|
||||
shown_api_base: Final = _api_base_without_credentials(api_base) if api_base else ""
|
||||
head: Final = f"\nModel: {model}" + (f"\nAPI Base: `{shown_api_base}`" if shown_api_base else "")
|
||||
prompt: Final = (
|
||||
f"\nMessages: `{messages}`" if messages and litellm.redact_messages_in_exceptions is False else ""
|
||||
)
|
||||
labelled_values: Final = (
|
||||
("model_group", metadata.get("model_group")),
|
||||
("deployment", metadata.get("deployment")),
|
||||
("vertex_project", params.get("vertex_project")),
|
||||
("vertex_location", params.get("vertex_location")),
|
||||
)
|
||||
tail: Final = "".join(f"\n{label}: `{value}`\n" for label, value in labelled_values if value is not None)
|
||||
call_id: Final = params.get("litellm_call_id")
|
||||
return _RequestDebugInfo(
|
||||
with_prompt=add_key_name_and_team_to_alert(request_info=head + prompt + tail, metadata=metadata),
|
||||
without_prompt=add_key_name_and_team_to_alert(request_info=head + tail, metadata=metadata),
|
||||
call_id=call_id if isinstance(call_id, str) else "",
|
||||
)
|
||||
except Exception:
|
||||
return _NO_REQUEST_DEBUG_INFO
|
||||
|
||||
|
||||
def _attach_debug_info(exception: Exception, request_debug_info: _RequestDebugInfo) -> None:
|
||||
attached_call_id: Final = getattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, None)
|
||||
if attached_call_id is _RAISED_BY_SEVERAL_CALLS:
|
||||
return
|
||||
if (
|
||||
isinstance(attached_call_id, str)
|
||||
and attached_call_id
|
||||
and request_debug_info.call_id
|
||||
and attached_call_id != request_debug_info.call_id
|
||||
):
|
||||
setattr(exception, _DEBUG_INFO_ATTRIBUTE, None)
|
||||
setattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, _RAISED_BY_SEVERAL_CALLS)
|
||||
return
|
||||
if request_debug_info.without_prompt and getattr(exception, _DEBUG_INFO_ATTRIBUTE, None) is None:
|
||||
setattr(exception, _DEBUG_INFO_ATTRIBUTE, request_debug_info.without_prompt)
|
||||
setattr(exception, _DEBUG_INFO_CALL_ID_ATTRIBUTE, request_debug_info.call_id)
|
||||
|
||||
|
||||
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:
|
||||
if not (model or custom_llm_provider):
|
||||
return error
|
||||
request_debug_info: Final = _request_debug_info(model, with_deployment_api_base(litellm_params), _NO_LITELLM_PARAMS)
|
||||
_attach_debug_info(error, request_debug_info)
|
||||
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 ""
|
||||
|
|
@ -2546,11 +2645,17 @@ def exception_type(
|
|||
model,
|
||||
original_exception,
|
||||
custom_llm_provider,
|
||||
completion_kwargs={},
|
||||
extra_kwargs={},
|
||||
completion_kwargs: object = _NO_LITELLM_PARAMS,
|
||||
extra_kwargs: object = _NO_LITELLM_PARAMS,
|
||||
):
|
||||
"""Maps an LLM Provider Exception to OpenAI Exception Format"""
|
||||
if any(isinstance(original_exception, exc_type) for exc_type in litellm.LITELLM_EXCEPTION_TYPES):
|
||||
request_debug_info: Final = (
|
||||
_request_debug_info(model, extra_kwargs, completion_kwargs)
|
||||
if model or custom_llm_provider
|
||||
else _NO_REQUEST_DEBUG_INFO
|
||||
)
|
||||
if isinstance(original_exception, tuple(litellm.LITELLM_EXCEPTION_TYPES)):
|
||||
_attach_debug_info(original_exception, request_debug_info)
|
||||
return original_exception
|
||||
if _is_guardrail_block(original_exception):
|
||||
return original_exception
|
||||
|
|
@ -2572,11 +2677,11 @@ def exception_type(
|
|||
print() # noqa: T201
|
||||
|
||||
litellm_response_headers: Final = _get_response_headers(original_exception=original_exception)
|
||||
extra_information: Final = request_debug_info.with_prompt
|
||||
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 = (
|
||||
|
|
@ -2589,47 +2694,8 @@ def exception_type(
|
|||
else:
|
||||
exception_type = ""
|
||||
|
||||
################################################################################
|
||||
# Common Extra information needed for all providers
|
||||
# We pass num retries, api_base, vertex_deployment etc to the exception here
|
||||
################################################################################
|
||||
try:
|
||||
_api_base: Final = litellm.get_api_base(model=model, optional_params=extra_kwargs)
|
||||
messages: Final = litellm.get_first_chars_messages(kwargs=completion_kwargs)
|
||||
_vertex_project: Final = extra_kwargs.get("vertex_project")
|
||||
_vertex_location: Final = extra_kwargs.get("vertex_location")
|
||||
_metadata: Final = extra_kwargs.get("metadata", {}) or {}
|
||||
_model_group: Final = _metadata.get("model_group")
|
||||
_deployment: Final = _metadata.get("deployment")
|
||||
extra_information = f"\nModel: {model}"
|
||||
|
||||
if isinstance(custom_llm_provider, str) and len(custom_llm_provider) > 0:
|
||||
exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception"
|
||||
|
||||
shown_api_base: Final = _api_base_without_credentials(_api_base) if _api_base else ""
|
||||
if shown_api_base:
|
||||
extra_information += f"\nAPI Base: `{shown_api_base}`"
|
||||
if messages and len(messages) > 0 and litellm.redact_messages_in_exceptions is False:
|
||||
extra_information += f"\nMessages: `{messages}`"
|
||||
|
||||
if _model_group is not None:
|
||||
extra_information += f"\nmodel_group: `{_model_group}`\n"
|
||||
if _deployment is not None:
|
||||
extra_information += f"\ndeployment: `{_deployment}`\n"
|
||||
if _vertex_project is not None:
|
||||
extra_information += f"\nvertex_project: `{_vertex_project}`\n"
|
||||
if _vertex_location is not None:
|
||||
extra_information += f"\nvertex_location: `{_vertex_location}`\n"
|
||||
|
||||
# on litellm proxy add key name + team to exceptions
|
||||
extra_information = add_key_name_and_team_to_alert(request_info=extra_information, metadata=_metadata)
|
||||
except Exception:
|
||||
# DO NOT LET this Block raising the original exception
|
||||
pass
|
||||
|
||||
################################################################################
|
||||
# End of Common Extra information Needed for all providers
|
||||
################################################################################
|
||||
if isinstance(custom_llm_provider, str) and len(custom_llm_provider) > 0:
|
||||
exception_provider = custom_llm_provider[0].upper() + custom_llm_provider[1:] + "Exception"
|
||||
|
||||
################################################################################
|
||||
#################### Start of Provider Exception mapping ####################
|
||||
|
|
@ -2923,17 +2989,20 @@ 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, request_debug_info)
|
||||
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, request_debug_info)
|
||||
raise e # it's already mapped
|
||||
raised_exc: Final = APIConnectionError(
|
||||
message=f"{original_exception}\n{redact_string(traceback.format_exc())}",
|
||||
llm_provider="",
|
||||
model="",
|
||||
)
|
||||
_attach_debug_info(raised_exc, request_debug_info)
|
||||
setattr(raised_exc, "litellm_response_headers", litellm_response_headers)
|
||||
carry_agentic_followup_mark(source=original_exception, mapped=raised_exc)
|
||||
raise raised_exc
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -563,6 +564,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:
|
||||
|
|
@ -610,8 +612,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.
|
||||
"""
|
||||
|
|
@ -802,18 +805,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
|
||||
|
|
|
|||
|
|
@ -90,6 +90,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 (
|
||||
AGENTIC_LOOP_CONTROL_KWARGS_KEYS,
|
||||
AGENTIC_LOOP_TOOL_CHOICE_KEY,
|
||||
|
|
@ -1033,7 +1034,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):
|
||||
|
|
@ -1042,11 +1043,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,
|
||||
)
|
||||
|
||||
|
||||
|
|
@ -8090,7 +8091,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,
|
||||
|
|
|
|||
|
|
@ -173,6 +173,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
|
||||
|
|
@ -271,6 +272,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(
|
||||
|
|
|
|||
|
|
@ -47,7 +47,7 @@ from litellm.litellm_core_utils.core_helpers import (
|
|||
set_response_cost_in_hidden_params,
|
||||
stamp_litellm_bill_on_usage,
|
||||
)
|
||||
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,
|
||||
|
|
@ -344,27 +344,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:
|
||||
|
|
@ -1219,12 +1213,17 @@ class BaseResponsesAPIStreamingIterator:
|
|||
model=self.model or "",
|
||||
)
|
||||
|
||||
def _with_deployment_debug_info(self, error: Exception) -> Exception:
|
||||
return with_deployment_debug_info(
|
||||
error, self.model or "", self.custom_llm_provider or "", _logged_litellm_params(self.logging_obj)
|
||||
)
|
||||
|
||||
def _raise_if_ended_without_terminal_event(self) -> None:
|
||||
if self.completed_response is not None:
|
||||
return
|
||||
if self._replays_stored_stream and not self._yielded_first_chunk:
|
||||
return
|
||||
error: Final = self._stream_ended_early_error()
|
||||
error: Final = self._with_deployment_debug_info(self._stream_ended_early_error())
|
||||
self._handle_failure(error)
|
||||
if self._output_started:
|
||||
raise error
|
||||
|
|
@ -1236,10 +1235,11 @@ class BaseResponsesAPIStreamingIterator:
|
|||
raise _mapped_exception(error, self.model or "", self.custom_llm_provider or "", self.logging_obj) from error
|
||||
|
||||
def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn:
|
||||
self._handle_failure(error)
|
||||
described: Final = self._with_deployment_debug_info(error)
|
||||
self._handle_failure(described)
|
||||
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(
|
||||
|
|
@ -1341,14 +1341,9 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
self.finished = True
|
||||
raise StopAsyncIteration from e
|
||||
self._raise_for_transport_timeout(e)
|
||||
except httpx.HTTPError as e:
|
||||
# Handle HTTP errors
|
||||
self.finished = True
|
||||
self._handle_failure(e)
|
||||
raise e
|
||||
except Exception as e:
|
||||
self.finished = True
|
||||
self._handle_failure(e)
|
||||
self._handle_failure(self._with_deployment_debug_info(e))
|
||||
raise e
|
||||
|
||||
def _handle_logging_completed_response(self):
|
||||
|
|
@ -1434,14 +1429,9 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
|
|||
self.finished = True
|
||||
raise StopIteration from e
|
||||
self._raise_for_transport_timeout(e)
|
||||
except httpx.HTTPError as e:
|
||||
# Handle HTTP errors
|
||||
self.finished = True
|
||||
self._handle_failure(e)
|
||||
raise e
|
||||
except Exception as e:
|
||||
self.finished = True
|
||||
self._handle_failure(e)
|
||||
self._handle_failure(self._with_deployment_debug_info(e))
|
||||
raise e
|
||||
|
||||
def _handle_logging_completed_response(self):
|
||||
|
|
|
|||
|
|
@ -6583,13 +6583,10 @@ def get_model_region(litellm_params: LiteLLM_Params, mode: str | None) -> str |
|
|||
return None
|
||||
|
||||
|
||||
def get_first_chars_messages(kwargs: dict) -> str:
|
||||
try:
|
||||
_messages = kwargs.get("messages")
|
||||
_messages = str(_messages)[:100]
|
||||
return _messages
|
||||
except Exception:
|
||||
return ""
|
||||
def get_first_chars_messages(kwargs: object) -> str:
|
||||
request: Final = as_str_mapping(kwargs)
|
||||
messages: Final = None if request is None else request.get("messages")
|
||||
return "" if messages is None else str(messages)[:100]
|
||||
|
||||
|
||||
def count_characters(text: str) -> int:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
@ -157,6 +158,265 @@ 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={"messages": [{"role": "user", "content": "CONFIDENTIAL patient record"}]},
|
||||
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 "CONFIDENTIAL" not in debug_info
|
||||
|
||||
|
||||
_ROUTER_METADATA: Final = {
|
||||
"api_base": "https://relay.example/v1",
|
||||
"model_group": "claude-alias",
|
||||
"deployment": "anthropic/claude-haiku-5-5",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("provider_error", "custom_llm_provider", "extra_kwargs"),
|
||||
[
|
||||
(
|
||||
AnthropicError(status_code=529, message="Overloaded"),
|
||||
"anthropic",
|
||||
{"litellm_metadata": _ROUTER_METADATA},
|
||||
),
|
||||
(
|
||||
OpenAIError(status_code=500, message="upstream failed"),
|
||||
"hosted_vllm",
|
||||
{
|
||||
"api_base": "https://relay.example/v1",
|
||||
"metadata": {"user_id": "set by the caller"},
|
||||
"litellm_metadata": _ROUTER_METADATA,
|
||||
},
|
||||
),
|
||||
],
|
||||
ids=["native_messages_without_api_base", "bridged_responses_with_request_metadata"],
|
||||
)
|
||||
def test_router_metadata_under_litellm_metadata_names_the_deployment(
|
||||
provider_error: Exception, custom_llm_provider: str, extra_kwargs: dict[str, object]
|
||||
) -> None:
|
||||
with pytest.raises(litellm.InternalServerError) as caught:
|
||||
exception_type(
|
||||
model="claude-haiku-5-5",
|
||||
original_exception=provider_error,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
extra_kwargs=extra_kwargs,
|
||||
)
|
||||
debug_info: Final = caught.value.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "API Base: `https://relay.example/v1`" in debug_info
|
||||
assert "model_group: `claude-alias`" in debug_info
|
||||
assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info
|
||||
|
||||
|
||||
_KEY_METADATA: Final = {"user_api_key_alias": "alias-key", "user_api_key_team_alias": "team-a"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"extra_kwargs",
|
||||
[
|
||||
{"metadata": _ROUTER_METADATA, "litellm_metadata": _KEY_METADATA},
|
||||
{"metadata": _KEY_METADATA, "litellm_metadata": _ROUTER_METADATA},
|
||||
],
|
||||
ids=["chat_with_guardrail_litellm_metadata", "generic_route_with_key_fields_in_metadata"],
|
||||
)
|
||||
def test_debug_info_reads_lines_from_both_metadata_buckets(extra_kwargs: dict[str, object]) -> None:
|
||||
with pytest.raises(litellm.InternalServerError) as caught:
|
||||
exception_type(
|
||||
model="claude-haiku-5-5",
|
||||
original_exception=AnthropicError(status_code=529, message="Overloaded"),
|
||||
custom_llm_provider="anthropic",
|
||||
extra_kwargs=extra_kwargs,
|
||||
)
|
||||
debug_info: Final = caught.value.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "Key Name: `alias-key`" in debug_info
|
||||
assert "Team: `team-a`" in debug_info
|
||||
assert "API Base: `https://relay.example/v1`" in debug_info
|
||||
assert "model_group: `claude-alias`" in debug_info
|
||||
assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info
|
||||
|
||||
|
||||
def test_litellm_metadata_wins_over_caller_metadata_for_the_same_key() -> None:
|
||||
with pytest.raises(litellm.InternalServerError) as caught:
|
||||
exception_type(
|
||||
model="claude-haiku-5-5",
|
||||
original_exception=AnthropicError(status_code=529, message="Overloaded"),
|
||||
custom_llm_provider="anthropic",
|
||||
extra_kwargs={
|
||||
"metadata": {"deployment": "set by the caller", "api_base": "https://caller.example"},
|
||||
"litellm_metadata": _ROUTER_METADATA,
|
||||
},
|
||||
)
|
||||
debug_info: Final = caught.value.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "deployment: `anthropic/claude-haiku-5-5`" in debug_info
|
||||
assert "API Base: `https://relay.example/v1`" in debug_info
|
||||
assert "caller" not in debug_info
|
||||
|
||||
|
||||
def test_mapper_that_passes_its_debug_info_keeps_the_prompt_line() -> None:
|
||||
request: Final = httpx.Request("POST", "https://gateway.example/v1/chat/completions")
|
||||
original: Final = OpenAIError(
|
||||
status_code=500, message="upstream failed", request=request, response=httpx.Response(500, request=request)
|
||||
)
|
||||
with pytest.raises(litellm.InternalServerError) as caught:
|
||||
exception_type(
|
||||
model="gpt-5.4-mini",
|
||||
original_exception=original,
|
||||
custom_llm_provider="openai",
|
||||
completion_kwargs={"messages": [{"role": "user", "content": "hello there"}]},
|
||||
extra_kwargs={"metadata": {"deployment": "openai/gpt-5.4-mini"}},
|
||||
)
|
||||
debug_info: Final = caught.value.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "Messages: `[{'role': 'user', 'content': 'hello there'}]`" in debug_info
|
||||
assert "deployment: `openai/gpt-5.4-mini`" in debug_info
|
||||
|
||||
|
||||
def test_debug_info_has_no_messages_line_when_the_request_has_no_messages() -> None:
|
||||
assert "Messages" not in _debug_info_for_api_base("https://gateway.example/v1")
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"already_mapped",
|
||||
[
|
||||
litellm.InternalServerError(
|
||||
message="Upstream generation aborted (finish_reason=abort)", model="gpt-5.4-mini", llm_provider="openai"
|
||||
),
|
||||
litellm.Timeout(
|
||||
message="Stream exceeded max streaming duration of 10s", model="gpt-5.4-mini", llm_provider="openai"
|
||||
),
|
||||
],
|
||||
ids=["finish_reason_abort", "max_streaming_duration"],
|
||||
)
|
||||
def test_already_mapped_error_gains_the_deployment_debug_info(already_mapped: Exception) -> None:
|
||||
returned: Final = exception_type(
|
||||
model="gpt-5.4-mini",
|
||||
original_exception=already_mapped,
|
||||
custom_llm_provider="openai",
|
||||
completion_kwargs={"messages": [{"role": "user", "content": "CONFIDENTIAL patient record"}]},
|
||||
extra_kwargs={
|
||||
"api_base": "https://relay.example/v1",
|
||||
"metadata": {"model_group": "gpt-alias", "deployment": "openai/gpt-5.4-mini"},
|
||||
},
|
||||
)
|
||||
assert returned is already_mapped
|
||||
debug_info: Final = returned.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "API Base: `https://relay.example/v1`" in debug_info
|
||||
assert "deployment: `openai/gpt-5.4-mini`" in debug_info
|
||||
assert "CONFIDENTIAL" not in debug_info
|
||||
|
||||
|
||||
def test_error_instance_raised_by_several_requests_names_none_of_them() -> None:
|
||||
shared: Final = litellm.RateLimitError(message="mock rate limit", llm_provider="openai", model="gpt-5.4-mini")
|
||||
|
||||
def debug_info_for(alias: str) -> object:
|
||||
with pytest.raises(litellm.RateLimitError) as caught:
|
||||
completion(
|
||||
model="openai/gpt-5.4-mini",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
mock_response=shared,
|
||||
metadata={"user_api_key_alias": alias, "model_group": f"group-{alias}"},
|
||||
)
|
||||
assert caught.value is shared
|
||||
return caught.value.litellm_debug_info
|
||||
|
||||
first: Final = debug_info_for("alias-a")
|
||||
assert isinstance(first, str) and "Key Name: `alias-a`" in first
|
||||
assert debug_info_for("alias-b") is None
|
||||
assert debug_info_for("alias-c") is None
|
||||
|
||||
|
||||
_RELAY_DEPLOYMENT_KWARGS: Final = {
|
||||
"api_base": "https://relay.example/v1",
|
||||
"metadata": {"deployment": "openai/gpt-5.4-mini"},
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("first_kwargs", "second_kwargs"),
|
||||
[
|
||||
({"litellm_call_id": "call-1", **_RELAY_DEPLOYMENT_KWARGS}, {}),
|
||||
(_RELAY_DEPLOYMENT_KWARGS, {"litellm_call_id": "call-1"}),
|
||||
],
|
||||
ids=["second-mapping-has-no-request", "first-mapping-had-no-call-id"],
|
||||
)
|
||||
def test_mapping_again_for_the_same_request_keeps_the_deployment_already_named(
|
||||
first_kwargs: dict[str, object], second_kwargs: dict[str, object]
|
||||
) -> None:
|
||||
already_mapped: Final = litellm.InternalServerError(
|
||||
message="Upstream generation aborted (finish_reason=abort)", model="gpt-5.4-mini", llm_provider="openai"
|
||||
)
|
||||
exception_type(
|
||||
model="gpt-5.4-mini",
|
||||
original_exception=already_mapped,
|
||||
custom_llm_provider="openai",
|
||||
extra_kwargs=first_kwargs,
|
||||
)
|
||||
exception_type(
|
||||
model="gpt-5.4-mini",
|
||||
original_exception=already_mapped,
|
||||
custom_llm_provider="openai",
|
||||
completion_kwargs={},
|
||||
extra_kwargs=second_kwargs,
|
||||
)
|
||||
debug_info: Final = already_mapped.litellm_debug_info
|
||||
assert isinstance(debug_info, str)
|
||||
assert "API Base: `https://relay.example/v1`" in debug_info
|
||||
assert "deployment: `openai/gpt-5.4-mini`" in debug_info
|
||||
|
||||
|
||||
class _ErrorWhoseMessageCannotBeRead(Exception):
|
||||
@property
|
||||
def message(self) -> str:
|
||||
raise RuntimeError("message unavailable")
|
||||
|
||||
|
||||
def test_error_that_breaks_the_mapping_still_names_the_deployment() -> None:
|
||||
with pytest.raises(litellm.APIConnectionError) as caught:
|
||||
exception_type(
|
||||
model="claude-opus-5-5",
|
||||
original_exception=_ErrorWhoseMessageCannotBeRead("upstream failed"),
|
||||
custom_llm_provider="anthropic",
|
||||
completion_kwargs={},
|
||||
extra_kwargs={"api_base": "https://relay.example", "metadata": {"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 "deployment: `anthropic/claude-opus-5-5`" in debug_info
|
||||
|
||||
|
||||
# Test cases for is_error_str_context_window_exceeded
|
||||
# Tuple format: (error_message, expected_result)
|
||||
context_window_test_cases = [
|
||||
|
|
|
|||
|
|
@ -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():
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import functools
|
|||
import gc
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
import weakref
|
||||
from collections.abc import Callable, Mapping
|
||||
from datetime import datetime
|
||||
|
|
@ -367,6 +368,136 @@ 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
|
||||
|
||||
|
||||
class _FailureDebugInfoRecorder(CustomLogger):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.debug_info: object = None
|
||||
|
||||
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
self.debug_info = getattr(kwargs["exception"], "litellm_debug_info", None)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES)
|
||||
def test_sync_transport_cut_failure_callbacks_see_the_deployment(prefix):
|
||||
recorder: Final = _FailureDebugInfoRecorder()
|
||||
logging_obj: Final = _real_logging_obj(
|
||||
call_type="responses",
|
||||
litellm_params={
|
||||
"metadata": {
|
||||
"api_base": "https://relay.example/openai/v1",
|
||||
"model_group": "responses-alias",
|
||||
"deployment": "openai/gpt-4o-mini",
|
||||
}
|
||||
},
|
||||
callbacks=(recorder,),
|
||||
)
|
||||
iterator: Final = _make_sync_iterator(
|
||||
sse_events=prefix,
|
||||
logging_obj=logging_obj,
|
||||
trailing_error=httpx.ReadError("peer closed connection without sending complete message body"),
|
||||
)
|
||||
|
||||
with pytest.raises((httpx.ReadError, MidStreamFallbackError)):
|
||||
for _ in iterator:
|
||||
pass
|
||||
|
||||
assert isinstance(recorder.debug_info, str)
|
||||
assert "API Base: `https://relay.example/openai/v1`" in recorder.debug_info
|
||||
assert "model_group: `responses-alias`" in recorder.debug_info
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES)
|
||||
async def test_stream_ending_without_terminal_event_names_the_deployment(prefix):
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
logging_obj.model_call_details = {
|
||||
"litellm_params": {
|
||||
"metadata": {
|
||||
"api_base": "https://relay.example/openai/v1",
|
||||
"model_group": "responses-alias",
|
||||
"deployment": "openai/gpt-4o-mini",
|
||||
}
|
||||
}
|
||||
}
|
||||
iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj)
|
||||
|
||||
with pytest.raises((litellm.APIConnectionError, 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 "deployment: `openai/gpt-4o-mini`" in debug_info
|
||||
|
||||
|
||||
_DEPLOYMENT_LITELLM_PARAMS: Final = {
|
||||
"metadata": {
|
||||
"api_base": "https://relay.example/openai/v1",
|
||||
"model_group": "responses-alias",
|
||||
"deployment": "openai/gpt-4o-mini",
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
async def _drain_async(iterator: ResponsesAPIStreamingIterator) -> None:
|
||||
async for _ in iterator:
|
||||
pass
|
||||
|
||||
|
||||
async def _drain_sync(iterator: SyncResponsesAPIStreamingIterator) -> None:
|
||||
for _ in iterator:
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("make_iterator", "drain"), [(_make_iterator, _drain_async), (_make_sync_iterator, _drain_sync)]
|
||||
)
|
||||
async def test_max_streaming_duration_timeout_names_the_deployment(monkeypatch, make_iterator, drain):
|
||||
monkeypatch.setattr(streaming_module, "LITELLM_MAX_STREAMING_DURATION_SECONDS", 10.0)
|
||||
logging_obj: Final = _failure_tracking_logging_obj()
|
||||
logging_obj.model_call_details = {"litellm_params": _DEPLOYMENT_LITELLM_PARAMS}
|
||||
iterator: Final = make_iterator(sse_events=_PARTIAL_OUTPUT_EVENTS, logging_obj=logging_obj)
|
||||
iterator._stream_created_time = time.time() - 20
|
||||
|
||||
with pytest.raises(litellm.Timeout, match="max streaming duration") as exc_info:
|
||||
await drain(iterator)
|
||||
|
||||
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 "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):
|
||||
|
|
|
|||
|
|
@ -11850,6 +11850,11 @@ async def test_success_enqueue_is_tracked_without_starting_callbacks_synchronous
|
|||
assert logging_obj.model_call_details["has_logged_async_success"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("kwargs", [None, "not a mapping"], ids=["none", "string"])
|
||||
def test_get_first_chars_messages_returns_empty_text_for_input_that_is_not_a_mapping(kwargs: object) -> None:
|
||||
assert litellm.get_first_chars_messages(kwargs=kwargs) == ""
|
||||
|
||||
|
||||
def test_get_response_string_joins_the_text_parts_of_a_completed_responses_event() -> None:
|
||||
from litellm.types.llms.openai import ResponseCompletedEvent
|
||||
from litellm.utils import get_response_string
|
||||
|
|
|
|||
|
|
@ -2,12 +2,14 @@ from collections.abc import AsyncIterator, Iterator, Sequence
|
|||
from itertools import chain
|
||||
from typing import Final
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from openai.types.completion import Completion
|
||||
|
||||
from litellm.exceptions import APIConnectionError
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
|
||||
from litellm.main import _async_streaming
|
||||
from litellm.text_completions.streaming_iterator import TextCompletionStreamWrapper
|
||||
from litellm.types.utils import Choices, Delta, ModelResponseStream, StreamingChoices, TextCompletionResponse, Usage
|
||||
|
||||
|
|
@ -288,6 +290,33 @@ def test_sync_text_completion_stream_maps_upstream_failure_after_content() -> No
|
|||
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
|
||||
|
||||
|
||||
def _stream_with_trailing_usage_chunk() -> tuple[ModelResponseStream, ...]:
|
||||
return (
|
||||
ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(content="Hi"))]),
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue