From 04412dcb9354558ddc0e83469756029664f2de5d Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Tue, 19 May 2026 16:09:42 -0700 Subject: [PATCH] perf: address greptile review for anthropic streaming hot path - Bail to legacy in `_collapse_pure_text_chunks` when content_block_delta events from different block indexes are observed without an intervening flush. Anthropic sends blocks strictly sequentially, but defensive bail prevents silent text-merging if the protocol ever interleaves. - Replace leaf-class `__dict__` check for `async_post_call_streaming_hook` in `_callback_capabilities` with a function-identity comparison that walks the MRO. A vendor base class can carry the override and the registered class can add nothing else; before this PR the hook was unconditionally invoked, so an inherited-override miss would silently drop the hook on the streaming path. - Add unit tests for both behaviors. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../anthropic_passthrough_logging_handler.py | 18 ++++- litellm/proxy/utils.py | 17 ++++- ...t_anthropic_passthrough_logging_handler.py | 66 +++++++++++++++++++ .../test_proxy_logging_hook_detection.py | 22 +++++++ 4 files changed, 120 insertions(+), 3 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py index 9421da87c3e..3be26eb572d 100644 --- a/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py +++ b/litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py @@ -390,10 +390,24 @@ class AnthropicPassthroughLoggingHandler: return None if dtype != "text_delta": return None - if data.get("index") not in text_block_indexes: + cur_index = data.get("index") + if cur_index not in text_block_indexes: + return None + # Defensive: Anthropic sends blocks strictly sequentially + # (start/deltas/stop, then next block), so pending_text from + # block N must be flushed by content_block_stop before block + # N+1's deltas arrive. If we ever see a delta whose index + # disagrees with the current pending buffer, the stream is + # interleaved -- fall back to legacy rather than risk merging + # text from different blocks under a single index. + if ( + pending_text + and pending_index is not None + and cur_index != pending_index + ): return None saw_any_text_delta = True - pending_index = data.get("index") + pending_index = cur_index pending_text.append(delta.get("text") or "") elif etype == "ping": # Interior no-op; legacy maps it to an empty chunk. diff --git a/litellm/proxy/utils.py b/litellm/proxy/utils.py index 032ab6c63b2..14f7f411e41 100644 --- a/litellm/proxy/utils.py +++ b/litellm/proxy/utils.py @@ -1596,7 +1596,22 @@ class ProxyLogging: iterator_overrides.append((resolved, "override")) elif "apply_guardrail" in cls_attrs: iterator_overrides.append((resolved, "apply_guardrail")) - if "async_post_call_streaming_hook" in cls_attrs: + # Walk the MRO for ``async_post_call_streaming_hook`` rather than + # using the leaf-class ``__dict__`` check used by the other flags: + # before this PR the hook was unconditionally invoked, so a + # callback that inherits an override from an intermediate parent + # (e.g. a vendor base class providing the override, with the + # registered class adding nothing else) MUST still be detected. + # A leaf-class miss here would silently drop the inherited hook. + base_streaming_hook = CustomLogger.async_post_call_streaming_hook + cls_streaming_hook = getattr( + cls, + "async_post_call_streaming_hook", + base_streaming_hook, + ) + if getattr( + cls_streaming_hook, "__func__", cls_streaming_hook + ) is not getattr(base_streaming_hook, "__func__", base_streaming_hook): has_streaming_chunk_override = True if "async_pre_call_hook" in cls_attrs: has_pre_call_override = True diff --git a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py index d2b71eeba63..0a9e3031030 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/llm_provider_handlers/test_anthropic_passthrough_logging_handler.py @@ -977,3 +977,69 @@ class TestPureTextFastPathParity: assert collapsed is not None # 50 text deltas + 50 event markers + 1 ping collapse to far fewer. assert len(collapsed) < len(all_chunks) / 2 + + def test_collapse_returns_none_for_interleaved_block_indexes(self): + """ + Anthropic sends content blocks strictly sequentially (start/deltas/stop + for one, then the next). If a stream ever interleaves deltas across + block indexes, the fast path must bail to legacy rather than merge text + from different blocks under a single index. + """ + frames = [ + self._sse( + "message_start", + { + "type": "message_start", + "message": { + "id": "msg_abc", + "type": "message", + "role": "assistant", + "model": "claude-3-5-sonnet-20241022", + "content": [], + "stop_reason": None, + "stop_sequence": None, + "usage": {"input_tokens": 1, "output_tokens": 0}, + }, + }, + ), + self._sse( + "content_block_start", + { + "type": "content_block_start", + "index": 0, + "content_block": {"type": "text", "text": ""}, + }, + ), + self._sse( + "content_block_start", + { + "type": "content_block_start", + "index": 1, + "content_block": {"type": "text", "text": ""}, + }, + ), + # Interleave: delta for block 0, then delta for block 1, with no + # content_block_stop between them. + self._sse( + "content_block_delta", + { + "type": "content_block_delta", + "index": 0, + "delta": {"type": "text_delta", "text": "hello "}, + }, + ), + self._sse( + "content_block_delta", + { + "type": "content_block_delta", + "index": 1, + "delta": {"type": "text_delta", "text": "world"}, + }, + ), + self._sse("message_stop", {"type": "message_stop"}), + ] + all_chunks = list(self._to_all_chunks(frames)) + assert ( + AnthropicPassthroughLoggingHandler._collapse_pure_text_chunks(all_chunks) + is None + ) diff --git a/tests/test_litellm/proxy/test_proxy_logging_hook_detection.py b/tests/test_litellm/proxy/test_proxy_logging_hook_detection.py index 4aebcf40aa5..f5967030561 100644 --- a/tests/test_litellm/proxy/test_proxy_logging_hook_detection.py +++ b/tests/test_litellm/proxy/test_proxy_logging_hook_detection.py @@ -111,6 +111,28 @@ def test_callback_capabilities_captures_iterator_override(monkeypatch): assert kind == "override" +def test_callback_capabilities_detects_inherited_streaming_chunk_override(monkeypatch): + """ + ``async_post_call_streaming_hook`` must be detected even when the override + lives on an intermediate parent class — a vendor base class can carry the + override and the registered class can add nothing else. Before this PR the + hook was unconditionally invoked, so a leaf-class ``__dict__`` miss here + would silently drop the inherited hook. + """ + ProxyLogging._callback_capabilities_cache.clear() + + class _StreamingBase(CustomLogger): + async def async_post_call_streaming_hook(self, *args, **kwargs): # type: ignore[override] + return kwargs.get("response") + + class _LeafWithoutOverride(_StreamingBase): + pass + + monkeypatch.setattr(litellm, "callbacks", [_LeafWithoutOverride()]) + caps = ProxyLogging._callback_capabilities() + assert caps.has_streaming_chunk_override is True + + def test_callback_capabilities_cache_invalidates_on_list_change(monkeypatch): """The cache key includes (length, id-of-each-callback). Mutating the callback list must produce a fresh capability snapshot."""