mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-15 04:19:08 +00:00
fix(logging): stop dropping streaming /v1/messages success logs
Two silent-loss paths for anthropic_messages streaming success logs: - get_standard_logging_object_payload swallows exceptions and returns None; loggers gated on standard_logging_object (e.g. s3_v2) then skip the event while spend tracking still succeeds via the response_cost fallback, so requests disappear from success logs with no error. Rebuild the payload from an empty response object when construction fails for an assembled streaming response, mirroring the failure-path fallback. Fixes #32019. - _handle_streaming_logging scheduled end-of-stream logging with an unreferenced asyncio.create_task, which can be garbage-collected before it runs. Enqueue on the logging worker instead, matching the fix applied to pass-through routes in #31485.
This commit is contained in:
parent
b4bb2a77a2
commit
327d4e20e1
3 changed files with 154 additions and 0 deletions
|
|
@ -3346,6 +3346,29 @@ class Logging(LiteLLMLoggingBaseClass):
|
|||
"LiteLLM.LoggingError: [Non-Blocking] Exception building the standard logging payload "
|
||||
"for a streaming response; callbacks still run without it"
|
||||
)
|
||||
if self.model_call_details.get("standard_logging_object") is None:
|
||||
# get_standard_logging_object_payload swallows exceptions and
|
||||
# returns None, and the except branch above leaves the key
|
||||
# unset. Loggers gated on standard_logging_object (e.g. s3_v2)
|
||||
# would then silently drop the request while spend tracking
|
||||
# still succeeds via the response_cost fallback. Rebuild from an
|
||||
# empty response object instead, mirroring the failure-path
|
||||
# fallback. See #32019.
|
||||
verbose_logger.error(
|
||||
"standard_logging_object build failed for streaming call_id=%s call_type=%s; "
|
||||
"rebuilding a degraded payload so success loggers still fire",
|
||||
self.litellm_call_id,
|
||||
self.call_type,
|
||||
)
|
||||
try:
|
||||
self.model_call_details["standard_logging_object"] = self._build_standard_logging_payload(
|
||||
{}, start_time, end_time
|
||||
)
|
||||
except Exception: # noqa: BLE001 # same rule as above: never block later callbacks
|
||||
verbose_logger.exception(
|
||||
"LiteLLM.LoggingError: [Non-Blocking] Exception rebuilding the degraded standard "
|
||||
"logging payload for a streaming response"
|
||||
)
|
||||
|
||||
# print standard logging payload
|
||||
if (standard_logging_payload := self.model_call_details.get("standard_logging_object")) is not None:
|
||||
|
|
|
|||
|
|
@ -0,0 +1,94 @@
|
|||
"""
|
||||
Regression tests for https://github.com/BerriAI/litellm/issues/32019.
|
||||
|
||||
Streaming /v1/messages responses were silently dropped by loggers gated on
|
||||
``standard_logging_object`` (e.g. ``s3_v2``) whenever StandardLoggingPayload
|
||||
construction failed: ``get_standard_logging_object_payload`` swallows the
|
||||
exception and returns ``None``, spend tracking still succeeds via the
|
||||
``response_cost`` fallback, and no error surfaces to the caller.
|
||||
"""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging
|
||||
from litellm.types.utils import ModelResponse
|
||||
|
||||
DEGRADED_PAYLOAD = {"trace_id": "degraded-fallback"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_slo_build_failure_falls_back_to_degraded_payload(monkeypatch):
|
||||
"""When the SLP build returns None for an assembled streaming response,
|
||||
async_success_handler must rebuild it from an empty response object so
|
||||
SLP-gated loggers still record the request."""
|
||||
logging_obj = Logging(
|
||||
model="claude-3-5-sonnet-20240620",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=True,
|
||||
call_type="anthropic_messages",
|
||||
start_time=datetime.now(),
|
||||
litellm_call_id="test-32019",
|
||||
function_id="test-32019",
|
||||
)
|
||||
build_calls = []
|
||||
|
||||
def fake_build(self, init_response_obj, start_time, end_time):
|
||||
build_calls.append(init_response_obj)
|
||||
if init_response_obj == {}:
|
||||
return DEGRADED_PAYLOAD
|
||||
return None # simulates get_standard_logging_object_payload swallowing an exception
|
||||
|
||||
monkeypatch.setattr(Logging, "_build_standard_logging_payload", fake_build)
|
||||
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
||||
monkeypatch.setattr(litellm, "success_callback", [])
|
||||
|
||||
result = ModelResponse(model="claude-3-5-sonnet-20240620")
|
||||
|
||||
await logging_obj.async_success_handler(
|
||||
result,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
)
|
||||
|
||||
assert logging_obj.model_call_details["standard_logging_object"] == DEGRADED_PAYLOAD
|
||||
assert build_calls == [result, {}]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_slo_build_success_does_not_rebuild(monkeypatch):
|
||||
"""The degraded rebuild must not run when the SLP builds normally."""
|
||||
logging_obj = Logging(
|
||||
model="claude-3-5-sonnet-20240620",
|
||||
messages=[{"role": "user", "content": "hi"}],
|
||||
stream=True,
|
||||
call_type="anthropic_messages",
|
||||
start_time=datetime.now(),
|
||||
litellm_call_id="test-32019-ok",
|
||||
function_id="test-32019-ok",
|
||||
)
|
||||
build_calls = []
|
||||
healthy_payload = {"trace_id": "healthy"}
|
||||
|
||||
def fake_build(self, init_response_obj, start_time, end_time):
|
||||
build_calls.append(init_response_obj)
|
||||
return healthy_payload
|
||||
|
||||
monkeypatch.setattr(Logging, "_build_standard_logging_payload", fake_build)
|
||||
monkeypatch.setattr(litellm, "_async_success_callback", [])
|
||||
monkeypatch.setattr(litellm, "success_callback", [])
|
||||
|
||||
result = ModelResponse(model="claude-3-5-sonnet-20240620")
|
||||
|
||||
await logging_obj.async_success_handler(
|
||||
result,
|
||||
start_time=datetime.now(),
|
||||
end_time=datetime.now(),
|
||||
cache_hit=False,
|
||||
)
|
||||
|
||||
assert logging_obj.model_call_details["standard_logging_object"] == healthy_payload
|
||||
assert len(build_calls) == 1
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
"""
|
||||
Regression test for https://github.com/BerriAI/litellm/issues/32019.
|
||||
|
||||
End-of-stream logging for /v1/messages must be enqueued on the logging
|
||||
worker instead of a bare ``asyncio.create_task``: an unreferenced task can
|
||||
be garbage-collected before it runs, dropping the success log entirely
|
||||
(same class of bug fixed for pass-through routes in #31485).
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
|
||||
from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
|
||||
BaseAnthropicMessagesStreamingIterator,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_streaming_logging_enqueues_on_logging_worker(monkeypatch):
|
||||
enqueued = []
|
||||
|
||||
def fake_enqueue(async_coroutine):
|
||||
enqueued.append(async_coroutine)
|
||||
async_coroutine.close()
|
||||
|
||||
monkeypatch.setattr(GLOBAL_LOGGING_WORKER, "ensure_initialized_and_enqueue", fake_enqueue)
|
||||
|
||||
iterator = BaseAnthropicMessagesStreamingIterator(
|
||||
litellm_logging_obj=MagicMock(),
|
||||
request_body={"model": "claude-3-5-sonnet-20240620", "stream": True},
|
||||
)
|
||||
|
||||
await iterator._handle_streaming_logging(collected_chunks=[b"data: {}\n\n"])
|
||||
|
||||
assert len(enqueued) == 1
|
||||
Loading…
Add table
Reference in a new issue