scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 49

fix(proxy): handle non-standard SSE frames in Anthropic passthrough logging (#26000)

Browse source 
Some third-party Anthropic-compatible providers emit non-standard SSE
frames (OpenAI-style [DONE] sentinels, non-JSON keep-alive lines) in
streaming responses. These caused json.JSONDecodeError in
_build_complete_streaming_response, breaking the passthrough logging
pipeline so the request was never logged or billed.

Skip whole-line 'data: [DONE]' sentinels and catch JSONDecodeError per
event. Matching the full line (not a substring) keeps a valid chunk
whose text payload contains '[DONE]' from being dropped.

Co-authored-by: shin-berri <shin-laptop@berri.ai>
Co-authored-by: yuneng-jiang <yuneng@berri.ai>
Co-authored-by: Sameer Kankute <sameer@berri.ai>
This commit is contained in:
Kelvin 2026-06-11 19:50:33 +08:00 • committed by GitHub
parent d13474800f
commit 8dfbe86fb1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 81 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

15
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
View file

@ -468,6 +468,13 @@ class AnthropicPassthroughLoggingHandler:
# Process each individual event
for event_str in individual_events:
try:
# Skip OpenAI-style [DONE] sentinels some Anthropic-compatible
# providers emit. Match the whole SSE line so a valid chunk whose
# text payload happens to contain "[DONE]" is not dropped.
if any(
line.strip() == "data: [DONE]" for line in event_str.split("\n")
):
continue
transformed_openai_chunk = anthropic_model_response_iterator.convert_str_chunk_to_generic_chunk(
chunk=event_str
)
@ -476,6 +483,14 @@ class AnthropicPassthroughLoggingHandler:
except (StopIteration, StopAsyncIteration):
break
except json.JSONDecodeError:
# Some upstreams emit non-JSON SSE lines; skip them so the
# logging pipeline is not broken by a single bad frame.
verbose_proxy_logger.debug(
"Skipping non-JSON SSE event: %s",
event_str[:200],
)
continue
complete_streaming_response = litellm.stream_chunk_builder(
chunks=all_openai_chunks,

66
tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py
View file

@ -724,6 +724,72 @@ class TestAnthropicBatchPassthroughCostTracking:
)
class TestBuildCompleteStreamingResponseRobustness:
"""_build_complete_streaming_response must tolerate non-standard SSE frames."""
def _build(self, chunks: List[str]):
return AnthropicPassthroughLoggingHandler._build_complete_streaming_response(
all_chunks=chunks,
litellm_logging_obj=MagicMock(),
model="claude-3-sonnet-20240229",
)
def test_done_frame_is_skipped(self):
"""A bare 'data: [DONE]' control frame must not break reconstruction."""
chunks = [
'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hi"}}',
'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
'event: message_stop\ndata: {"type":"message_stop"}',
"data: [DONE]",
]
result = self._build(chunks)
assert result is not None
assert result.choices[0].message.content == "Hi"
def test_non_json_sse_line_is_skipped(self):
"""Non-JSON SSE lines (comments, keep-alive pings) must be skipped."""
chunks = [
": ping",
'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
"this is not json at all",
]
# Must not raise; a malformed stream simply yields no usable response.
result = self._build(chunks)
assert result is None or hasattr(result, "choices")
def test_mixed_valid_and_invalid_frames(self):
"""Valid events are still collected when interleaved with invalid ones."""
chunks = [
'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
"data: [DONE]",
": keep-alive",
"not-json",
'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}',
'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":2}}',
'event: message_stop\ndata: {"type":"message_stop"}',
]
result = self._build(chunks)
assert result is not None
assert result.choices[0].message.content == "Hello"
def test_done_in_text_payload_is_not_dropped(self):
"""A valid event whose text content contains '[DONE]' must NOT be skipped."""
chunks = [
'event: message_start\ndata: {"type":"message_start","message":{"id":"msg_1","type":"message","role":"assistant","content":[],"model":"claude-3-sonnet-20240229","stop_reason":null,"stop_sequence":null,"usage":{"input_tokens":10,"output_tokens":1}}}',
'event: content_block_start\ndata: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}',
'event: content_block_delta\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"The stream ends with [DONE]"}}',
'event: content_block_stop\ndata: {"type":"content_block_stop","index":0}',
'event: message_delta\ndata: {"type":"message_delta","delta":{"stop_reason":"end_turn","stop_sequence":null},"usage":{"output_tokens":8}}',
'event: message_stop\ndata: {"type":"message_stop"}',
]
result = self._build(chunks)
assert result is not None
assert result.choices[0].message.content == "The stream ends with [DONE]"
class TestPureTextFastPathParity:
"""
The pure-text fast path in _build_complete_streaming_response must produce