mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
fix(passthrough): skip [DONE] sentinels and non-JSON SSE frames in Anthropic streaming logging
Targeted subset of staging commitcfcdf8714a(#30202): only the anthropic_passthrough_logging_handler.py hardening hunks and their four tests are taken; the rest of that staging batch is intentionally excluded. (cherry picked from commitcfcdf8714a) (cherry picked from commit973c7eb8d6)
This commit is contained in:
parent
f2da26e340
commit
a0b14316c5
2 changed files with 81 additions and 0 deletions
|
|
@ -350,6 +350,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
|
||||
)
|
||||
|
|
@ -358,6 +365,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,
|
||||
|
|
|
|||
|
|
@ -949,6 +949,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 TestStreamFalseDeduplication:
|
||||
"""
|
||||
Regression tests for the duplicate-callback bug where a streaming pass-through
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue