mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
Merge 114678a073 into e9b1c11098
This commit is contained in:
Sandiyo Christan
GitHub
commit
8a4c045c10
3 changed files with 174 additions and 15 deletions
|
|
@ -7394,6 +7394,15 @@ class StandardLoggingPayloadSetup:
|
|||
# ProxyException uses .code, LiteLLM exceptions use .status_code,
|
||||
# httpx.HTTPStatusError exposes status only as .response.status_code.
|
||||
# Stringified for Prisma JSON compatibility.
|
||||
# If wrapped in MidStreamFallbackError (or similar wrapper carrying .original_exception),
|
||||
# unwrap to the carried provider/mapped exception so error_class, status_code, and provider
|
||||
# reflect the underlying failure rather than the internal wrapper.
|
||||
current_exc: Exception | None = original_exception
|
||||
while getattr(current_exc, "original_exception", None) is not None:
|
||||
current_exc = getattr(current_exc, "original_exception")
|
||||
if current_exc is not None:
|
||||
original_exception = current_exc
|
||||
|
||||
error_code_attr: Final[object] = getattr(original_exception, "code", None)
|
||||
if error_code_attr is not None and str(error_code_attr) not in ("", "None"):
|
||||
error_status: str = str(error_code_attr)
|
||||
|
|
|
|||
|
|
@ -2508,10 +2508,11 @@ class CustomStreamWrapper:
|
|||
return processed_chunk
|
||||
except Exception as e:
|
||||
traceback_exception: Final = traceback.format_exc()
|
||||
mapped_exception: Exception = self._get_mapped_stream_exception(e)
|
||||
# LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
|
||||
threading.Thread(target=self.logging_obj.failure_handler, args=(e, traceback_exception)).start()
|
||||
threading.Thread(target=self.logging_obj.failure_handler, args=(mapped_exception, traceback_exception)).start()
|
||||
self.close()
|
||||
self._handle_stream_fallback_error(e)
|
||||
self._handle_stream_fallback_error(e, mapped_exception=mapped_exception)
|
||||
|
||||
def fetch_sync_stream(self):
|
||||
if self.completion_stream is None and self.make_call is not None:
|
||||
|
|
@ -2628,9 +2629,10 @@ class CustomStreamWrapper:
|
|||
traceback_exception = traceback.format_exc()
|
||||
## ADD DEBUG INFORMATION - E.G. LITELLM REQUEST TIMEOUT
|
||||
traceback_exception += f"\nLiteLLM Default Request Timeout - {litellm.request_timeout}"
|
||||
self._dispatch_stream_failure(e, traceback_exception)
|
||||
mapped_exception = self._get_mapped_stream_exception(e)
|
||||
self._dispatch_stream_failure(mapped_exception, traceback_exception)
|
||||
await self.aclose()
|
||||
self._handle_stream_fallback_error(e)
|
||||
self._handle_stream_fallback_error(e, mapped_exception=mapped_exception)
|
||||
except (httpx.ReadError, httpx.RemoteProtocolError) as e:
|
||||
if self.received_finish_reason is None:
|
||||
await self.aclose()
|
||||
|
|
@ -2731,8 +2733,9 @@ class CustomStreamWrapper:
|
|||
return processed_chunk
|
||||
|
||||
def _log_stream_failure_and_raise(self, e: Exception) -> NoReturn:
|
||||
self._dispatch_stream_failure(e, traceback.format_exc())
|
||||
self._handle_stream_fallback_error(e)
|
||||
mapped_exception = self._get_mapped_stream_exception(e)
|
||||
self._dispatch_stream_failure(mapped_exception, traceback.format_exc())
|
||||
self._handle_stream_fallback_error(e, mapped_exception=mapped_exception)
|
||||
|
||||
def _dispatch_stream_failure(self, e: Exception, traceback_exception: str) -> None:
|
||||
self._record_partial_usage_for_failure()
|
||||
|
|
@ -2854,7 +2857,19 @@ class CustomStreamWrapper:
|
|||
extra_kwargs=with_deployment_api_base(self.logging_obj.model_call_details.get("litellm_params")),
|
||||
)
|
||||
|
||||
def _handle_stream_fallback_error(self, e: Exception) -> "NoReturn":
|
||||
def _get_mapped_stream_exception(self, e: Exception) -> Exception:
|
||||
from litellm.exceptions import MidStreamFallbackError
|
||||
|
||||
if isinstance(e, MidStreamFallbackError):
|
||||
return e.original_exception if e.original_exception is not None else e
|
||||
if isinstance(e, OpenAIError):
|
||||
return e
|
||||
try:
|
||||
return self._map_stream_exception(e)
|
||||
except Exception as mapping_error:
|
||||
return mapping_error
|
||||
|
||||
def _handle_stream_fallback_error(self, e: Exception, mapped_exception: Exception | None = None) -> "NoReturn":
|
||||
"""
|
||||
Common error handling for both __next__ and __anext__.
|
||||
|
||||
|
|
@ -2890,19 +2905,20 @@ class CustomStreamWrapper:
|
|||
emitted_disqualifying_content=self._emitted_disqualifying_content or e.emitted_disqualifying_content,
|
||||
)
|
||||
|
||||
# Map to OpenAI exception format. Some providers' mappers (e.g.
|
||||
# Map to OpenAI exception format if not already mapped. Some providers' mappers (e.g.
|
||||
# _map_anthropic_exception, _map_aleph_alpha_exception) synchronously
|
||||
# log a debug diagnostic (the raw status code) as part of mapping -
|
||||
# restore the consumer's outer context only after this completes, so
|
||||
# that diagnostic log line still carries the failing stream's own
|
||||
# trace_id/session_id instead of the consumer's (or an empty one).
|
||||
if isinstance(e, OpenAIError):
|
||||
mapped_exception: Exception = e
|
||||
else:
|
||||
try:
|
||||
mapped_exception = self._map_stream_exception(e)
|
||||
except Exception as mapping_error:
|
||||
mapped_exception = mapping_error
|
||||
if mapped_exception is None:
|
||||
if isinstance(e, OpenAIError):
|
||||
mapped_exception = e
|
||||
else:
|
||||
try:
|
||||
mapped_exception = self._map_stream_exception(e)
|
||||
except Exception as mapping_error:
|
||||
mapped_exception = mapping_error
|
||||
self._restore_consumer_correlation_context()
|
||||
|
||||
def _normalize_status_code(exc: Exception) -> int | None:
|
||||
|
|
|
|||
134
tests/unit/test_stream_failure_logging_error_class.py
Normal file
134
tests/unit/test_stream_failure_logging_error_class.py
Normal file
|
|
@ -0,0 +1,134 @@
|
|||
import asyncio
|
||||
import threading
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.exceptions import InternalServerError, MidStreamFallbackError
|
||||
from litellm.integrations.custom_logger import CustomLogger
|
||||
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
|
||||
|
||||
|
||||
class _MockSSEHandler(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
self.rfile.read(int(self.headers.get("content-length", 0)))
|
||||
self.send_response(200)
|
||||
self.send_header("content-type", "text/event-stream")
|
||||
self.end_headers()
|
||||
sse_payload = (
|
||||
'event: error\ndata: {"type": "error", "error": {"type": "overloaded_error", "message": "Overloaded"}}\n\n'
|
||||
)
|
||||
self.wfile.write(sse_payload.encode())
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_sse_server():
|
||||
server = HTTPServer(("127.0.0.1", 0), _MockSSEHandler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
yield server
|
||||
server.shutdown()
|
||||
|
||||
|
||||
def test_get_error_information_unwraps_mid_stream_fallback_error():
|
||||
inner = InternalServerError(
|
||||
message="Anthropic overloaded",
|
||||
response=None,
|
||||
llm_provider="anthropic",
|
||||
model="claude-sonnet-4-5",
|
||||
)
|
||||
wrapper = MidStreamFallbackError(
|
||||
message="MidStreamFallbackError: overloaded",
|
||||
original_exception=inner,
|
||||
model="claude-sonnet-4-5",
|
||||
llm_provider="anthropic",
|
||||
)
|
||||
|
||||
info = StandardLoggingPayloadSetup.get_error_information(wrapper)
|
||||
assert info["error_class"] == "InternalServerError"
|
||||
assert info["error_code"] == "500"
|
||||
assert info["llm_provider"] == "anthropic"
|
||||
|
||||
|
||||
def test_stream_failure_callback_logs_mapped_error_class_sync(mock_sse_server):
|
||||
logged = []
|
||||
|
||||
class _SyncFailLogger(CustomLogger):
|
||||
def log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
std_obj = kwargs.get("standard_logging_object", {})
|
||||
if "error_information" in std_obj:
|
||||
logged.append(std_obj["error_information"])
|
||||
|
||||
original_callbacks = list(litellm.callbacks)
|
||||
litellm.callbacks = [_SyncFailLogger()]
|
||||
try:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
for _ in litellm.completion(
|
||||
model="anthropic/claude-sonnet-4-5",
|
||||
api_key="sk-mock-key",
|
||||
api_base=f"http://127.0.0.1:{mock_sse_server.server_address[1]}",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=True,
|
||||
):
|
||||
pass
|
||||
|
||||
# Underlying exception raised/wrapped should be InternalServerError
|
||||
raised = getattr(exc_info.value, "original_exception", None) or exc_info.value
|
||||
assert isinstance(raised, InternalServerError)
|
||||
|
||||
for _ in range(50):
|
||||
if logged:
|
||||
break
|
||||
time.sleep(0.05)
|
||||
|
||||
assert len(logged) > 0
|
||||
info = logged[0]
|
||||
assert info["error_class"] == "InternalServerError"
|
||||
assert str(info["error_code"]) == "500"
|
||||
finally:
|
||||
litellm.callbacks = original_callbacks
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_failure_callback_logs_mapped_error_class_async(mock_sse_server):
|
||||
logged = []
|
||||
|
||||
class _AsyncFailLogger(CustomLogger):
|
||||
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
|
||||
std_obj = kwargs.get("standard_logging_object", {})
|
||||
if "error_information" in std_obj:
|
||||
logged.append(std_obj["error_information"])
|
||||
|
||||
original_callbacks = list(litellm.callbacks)
|
||||
litellm.callbacks = [_AsyncFailLogger()]
|
||||
try:
|
||||
with pytest.raises(Exception) as exc_info:
|
||||
stream = await litellm.acompletion(
|
||||
model="anthropic/claude-sonnet-4-5",
|
||||
api_key="sk-mock-key",
|
||||
api_base=f"http://127.0.0.1:{mock_sse_server.server_address[1]}",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=True,
|
||||
)
|
||||
async for _ in stream:
|
||||
pass
|
||||
|
||||
raised = getattr(exc_info.value, "original_exception", None) or exc_info.value
|
||||
assert isinstance(raised, InternalServerError)
|
||||
|
||||
for _ in range(50):
|
||||
if logged:
|
||||
break
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
assert len(logged) > 0
|
||||
info = logged[0]
|
||||
assert info["error_class"] == "InternalServerError"
|
||||
assert str(info["error_code"]) == "500"
|
||||
finally:
|
||||
litellm.callbacks = original_callbacks
|
||||
Loading…
Add table
Reference in a new issue