scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-14 09:03:16 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 214

Merge a39d97b7e5 into fb74bc902f

Browse source
This commit is contained in:
Tanvir Alam 2026-11-05 20:15:19 +00:00 • committed by GitHub
commit 6105d65f06
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 20 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
litellm/caching/caching_handler.py
View file

@ -182,14 +182,16 @@ def _should_defer_streaming_cache_hit_callbacks(*, cached_result: object) -> boo
spend and callback records. A plain (non-stream) replay logs here, since nothing
else will.
"""
from litellm.llms.anthropic.messages.native.response_cache import (
CachedAnthropicMessagesStreamIterator,
)
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
return isinstance(
cached_result, (CustomStreamWrapper, BaseResponsesAPIStreamingIterator, CachedAnthropicMessagesStreamIterator)
)
streaming_types: Final[tuple[type, ...]] = (CustomStreamWrapper, BaseResponsesAPIStreamingIterator)
try:
from litellm.llms.anthropic.messages.native.response_cache import (
CachedAnthropicMessagesStreamIterator,
)
except ImportError:
return isinstance(cached_result, streaming_types)
return isinstance(cached_result, (*streaming_types, CachedAnthropicMessagesStreamIterator))
def _prompt_tokens_details_as_mapping(details: "PromptTokensDetailsWrapper") -> Mapping[str, object]:

12
tests/unit/caching/test_caching_handler.py
View file

@ -1,4 +1,5 @@
import time
import sys
import traceback
from litellm._uuid import uuid
@ -1281,6 +1282,17 @@ def test_should_defer_streaming_cache_hit_callbacks_for_any_streaming_request():
assert _should_defer_streaming_cache_hit_callbacks(cached_result={"id": "msg_1"}) is False
def test_should_defer_streaming_cache_hit_callbacks_without_proxy_extra(monkeypatch):
monkeypatch.setitem(sys.modules, "litellm.llms.anthropic.messages.native.response_cache", None)
logging_obj = MagicMock()
logging_obj.model_call_details = {}
stream_replay = CustomStreamWrapper(
completion_stream=iter(()), model="gpt-4o", logging_obj=logging_obj
)
assert _should_defer_streaming_cache_hit_callbacks(cached_result=stream_replay) is True
assert _should_defer_streaming_cache_hit_callbacks(cached_result=ModelResponse()) is False
@pytest.mark.asyncio
async def test_async_get_cache_defers_streaming_completion_hit_callbacks():
litellm.set_verbose = True