import time import traceback from litellm._uuid import uuid from dotenv import load_dotenv load_dotenv() import asyncio import hashlib import random import pytest import litellm from litellm import aembedding, completion, embedding, aresponses, responses from litellm.caching.caching import Cache from litellm.responses.streaming_iterator import CachedResponsesAPIStreamingIterator from unittest.mock import AsyncMock, patch, MagicMock from litellm.caching.caching_handler import ( LLMCachingHandler, CachingHandlerResponse, _is_chat_completion_cached_dict, _should_defer_streaming_cache_hit_callbacks, ) from litellm.caching.caching import LiteLLMCacheType from litellm.types.utils import CallTypes from litellm.types.rerank import RerankResponse from litellm.types.utils import ( ModelResponse, EmbeddingResponse, TextCompletionResponse, TranscriptionResponse, Embedding, ) from litellm.types.llms.openai import ResponsesAPIResponse from datetime import timedelta, datetime from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper from litellm._logging import verbose_logger import logging def setup_cache(): # Set up the cache cache = Cache(type=LiteLLMCacheType.LOCAL) litellm.cache = cache return cache chat_completion_response = litellm.ModelResponse( id=str(uuid.uuid4()), choices=[ litellm.Choices( message=litellm.Message( role="assistant", content="Hello, how can I help you today?" ) ) ], ) text_completion_response = litellm.TextCompletionResponse( id=str(uuid.uuid4()), choices=[litellm.utils.TextChoices(text="Hello, how can I help you today?")], ) @pytest.mark.asyncio @pytest.mark.parametrize( "response", [chat_completion_response, text_completion_response] ) async def test_async_set_get_cache(response): litellm.set_verbose = True setup_cache() verbose_logger.setLevel(logging.DEBUG) caching_handler = LLMCachingHandler( original_function=completion, request_kwargs={}, start_time=datetime.now() ) messages = [{"role": "user", "content": f"Unique message {datetime.now()}"}] logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.completion.value, model="gpt-3.5-turbo", messages=messages, function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) result = response print("result", result) original_function = ( litellm.acompletion if isinstance(response, litellm.ModelResponse) else litellm.atext_completion ) if isinstance(response, litellm.ModelResponse): kwargs = {"messages": messages} call_type = CallTypes.acompletion.value else: kwargs = {"prompt": f"Hello, how can I help you today? {datetime.now()}"} call_type = CallTypes.atext_completion.value await caching_handler.async_set_cache( result=result, original_function=original_function, kwargs=kwargs ) await asyncio.sleep(2) # Verify the result was cached cached_response = await caching_handler._async_get_cache( model="gpt-3.5-turbo", original_function=original_function, logging_obj=logging_obj, start_time=datetime.now(), call_type=call_type, kwargs=kwargs, ) assert cached_response.cached_result is not None assert cached_response.cached_result.id == result.id @pytest.mark.asyncio async def test_async_log_cache_hit_on_callbacks(): """ Assert logging callbacks are called after a cache hit """ # Setup caching_handler = LLMCachingHandler( original_function=completion, request_kwargs={}, start_time=datetime.now() ) mock_logging_obj = MagicMock() mock_logging_obj.async_success_handler = AsyncMock() mock_logging_obj.success_handler = MagicMock() mock_logging_obj.handle_sync_success_callbacks_for_async_calls = MagicMock() cached_result = "Mocked cached result" start_time = datetime.now() end_time = start_time + timedelta(seconds=1) cache_hit = True # Call the method caching_handler._async_log_cache_hit_on_callbacks( logging_obj=mock_logging_obj, cached_result=cached_result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, ) # Wait for the async task to complete await asyncio.sleep(0.5) print("mock logging obj methods called", mock_logging_obj.mock_calls) # Assertions mock_logging_obj.async_success_handler.assert_called_once_with( result=cached_result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, ) # Wait for the thread to complete await asyncio.sleep(0.5) mock_logging_obj.handle_sync_success_callbacks_for_async_calls.assert_called_once_with( result=cached_result, start_time=start_time, end_time=end_time, cache_hit=cache_hit, ) @pytest.mark.parametrize( "call_type, cached_result, expected_type", [ ( CallTypes.completion.value, { "id": "test", "choices": [{"message": {"role": "assistant", "content": "Hello"}}], }, ModelResponse, ), ( CallTypes.text_completion.value, {"id": "test", "choices": [{"text": "Hello"}]}, TextCompletionResponse, ), ( CallTypes.embedding.value, {"data": [{"embedding": [0.1, 0.2, 0.3]}]}, EmbeddingResponse, ), ( CallTypes.rerank.value, {"id": "test", "results": [{"index": 0, "relevance_score": 0.9}]}, RerankResponse, ), ( CallTypes.transcription.value, {"text": "Hello, world!"}, TranscriptionResponse, ), ], ) def test_convert_cached_result_to_model_response( call_type, cached_result, expected_type ): """ Assert that the cached result is converted to the correct type """ caching_handler = LLMCachingHandler( original_function=lambda: None, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=call_type, model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello, how can I help you today?"}], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=call_type, kwargs={}, logging_obj=logging_obj, model="test-model", args=(), ) assert isinstance(result, expected_type) assert result is not None def test_combine_cached_embedding_response_with_api_result(): """ If the cached response has [cache_hit, None, cache_hit] result should be [cache_hit, api_result, cache_hit] """ # Setup caching_handler = LLMCachingHandler( original_function=lambda: None, request_kwargs={}, start_time=datetime.now() ) start_time = datetime.now() end_time = start_time + timedelta(seconds=1) # Create a CachingHandlerResponse with some cached and some None values cached_response = EmbeddingResponse( data=[ Embedding(embedding=[0.1, 0.2, 0.3], index=0, object="embedding"), None, Embedding(embedding=[0.7, 0.8, 0.9], index=2, object="embedding"), ] ) caching_handler_response = CachingHandlerResponse( final_embedding_cached_response=cached_response ) # Create an API EmbeddingResponse for the missing value api_response = EmbeddingResponse( data=[Embedding(embedding=[0.4, 0.5, 0.6], index=1, object="embedding")] ) # Call the method result = caching_handler._combine_cached_embedding_response_with_api_result( _caching_handler_response=caching_handler_response, embedding_response=api_response, start_time=start_time, end_time=end_time, ) # Assertions assert isinstance(result, EmbeddingResponse) assert len(result.data) == 3 assert result.data[0].embedding == [0.1, 0.2, 0.3] assert result.data[1].embedding == [0.4, 0.5, 0.6] assert result.data[2].embedding == [0.7, 0.8, 0.9] assert result._hidden_params["cache_hit"] == True assert isinstance(result._response_ms, float) assert result._response_ms > 0 def test_combine_cached_embedding_response_multiple_missing_values(): """ If the cached response has [cache_hit, None, None, cache_hit, None] result should be [cache_hit, api_result, api_result, cache_hit, api_result] """ # Setup caching_handler = LLMCachingHandler( original_function=lambda: None, request_kwargs={}, start_time=datetime.now() ) start_time = datetime.now() end_time = start_time + timedelta(seconds=1) # Create a CachingHandlerResponse with some cached and some None values cached_response = EmbeddingResponse( data=[ Embedding(embedding=[0.1, 0.2, 0.3], index=0, object="embedding"), None, None, Embedding(embedding=[0.7, 0.8, 0.9], index=3, object="embedding"), None, ] ) caching_handler_response = CachingHandlerResponse( final_embedding_cached_response=cached_response ) # Create an API EmbeddingResponse for the missing values api_response = EmbeddingResponse( data=[ Embedding(embedding=[0.4, 0.5, 0.6], index=1, object="embedding"), Embedding(embedding=[0.4, 0.5, 0.6], index=2, object="embedding"), Embedding(embedding=[0.4, 0.5, 0.6], index=4, object="embedding"), ] ) # Call the method result = caching_handler._combine_cached_embedding_response_with_api_result( _caching_handler_response=caching_handler_response, embedding_response=api_response, start_time=start_time, end_time=end_time, ) # Assertions assert isinstance(result, EmbeddingResponse) assert len(result.data) == 5 assert result.data[0].embedding == [0.1, 0.2, 0.3] assert result.data[1].embedding == [0.4, 0.5, 0.6] assert result.data[2].embedding == [0.4, 0.5, 0.6] assert result.data[3].embedding == [0.7, 0.8, 0.9] @pytest.mark.asyncio async def test_embedding_cache_model_field_consistency(): """ Test that the model field is consistently preserved in cached embedding responses. This ensures that cache hits return the same model field as the original API response. """ # Setup cache setup_cache() caching_handler = LLMCachingHandler( original_function=aembedding, request_kwargs={}, start_time=datetime.now() ) # Create a mock embedding response with a specific model original_model = "text-embedding-005" embedding_response = EmbeddingResponse( model=original_model, data=[ Embedding(embedding=[0.1, 0.2, 0.3], index=0, object="embedding"), Embedding(embedding=[0.4, 0.5, 0.6], index=1, object="embedding"), ], ) # Mock logging object logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aembedding.value, model=original_model, messages=[], # Not used for embeddings function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) # Test parameters kwargs = { "model": original_model, "input": ["test input 1", "test input 2"], "caching": True, } # Step 1: Cache the embedding response await caching_handler.async_set_cache( result=embedding_response, original_function=aembedding, kwargs=kwargs ) # Step 2: Retrieve from cache cached_response = await caching_handler._async_get_cache( model=original_model, original_function=aembedding, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.aembedding.value, kwargs=kwargs, ) # Step 3: Verify the model field is preserved assert cached_response.final_embedding_cached_response is not None assert cached_response.final_embedding_cached_response.model == original_model assert len(cached_response.final_embedding_cached_response.data) == 2 assert cached_response.final_embedding_cached_response.data[0].embedding == [ 0.1, 0.2, 0.3, ] assert cached_response.final_embedding_cached_response.data[0].index == 0 assert cached_response.final_embedding_cached_response.data[1].embedding == [ 0.4, 0.5, 0.6, ] assert cached_response.final_embedding_cached_response.data[1].index == 1 # Verify cache hit flag is set assert ( cached_response.final_embedding_cached_response._hidden_params["cache_hit"] == True ) @pytest.mark.asyncio async def test_embedding_cache_model_field_with_vendor_prefix(): """ Test that the model field is preserved even when using vendor-prefixed model names. This simulates the real-world scenario where models might be prefixed with vendor names. """ # Setup cache setup_cache() caching_handler = LLMCachingHandler( original_function=aembedding, request_kwargs={}, start_time=datetime.now() ) # Test with vendor-prefixed model name (like vertex_ai/text-embedding-005) vendor_model = "vertex_ai/text-embedding-005" actual_model = "text-embedding-005" # What the provider actually returns # Create embedding response with the actual model name (as returned by provider) embedding_response = EmbeddingResponse( model=actual_model, # Provider returns this data=[ Embedding(embedding=[0.1, 0.2, 0.3], index=0, object="embedding"), ], ) # Mock logging object logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aembedding.value, model=vendor_model, messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) # Test parameters with vendor-prefixed model kwargs = { "model": vendor_model, # Request uses vendor prefix "input": ["test input"], "caching": True, } # Cache the response await caching_handler.async_set_cache( result=embedding_response, original_function=aembedding, kwargs=kwargs ) # Retrieve from cache cached_response = await caching_handler._async_get_cache( model=vendor_model, original_function=aembedding, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.aembedding.value, kwargs=kwargs, ) # Verify the model field matches the original provider response, not the request assert cached_response.final_embedding_cached_response is not None assert ( cached_response.final_embedding_cached_response.model == actual_model ) # Should be the provider's model name assert ( cached_response.final_embedding_cached_response.model != vendor_model ) # Should NOT be the vendor-prefixed name def test_extract_model_from_cached_results(): """ Test the helper method that extracts model names from cached results. """ caching_handler = LLMCachingHandler( original_function=aembedding, request_kwargs={}, start_time=datetime.now() ) # Test with valid cached results non_null_list = [ ( 0, { "embedding": [0.1, 0.2], "index": 0, "object": "embedding", "model": "text-embedding-005", }, ), ( 1, { "embedding": [0.3, 0.4], "index": 1, "object": "embedding", "model": "text-embedding-005", }, ), ] model_name = caching_handler._extract_model_from_cached_results(non_null_list) assert model_name == "text-embedding-005" # Test with missing model field non_null_list_no_model = [ (0, {"embedding": [0.1, 0.2], "index": 0, "object": "embedding"}), (1, {"embedding": [0.3, 0.4], "index": 1, "object": "embedding"}), ] model_name = caching_handler._extract_model_from_cached_results( non_null_list_no_model ) assert model_name is None # Test with empty list model_name = caching_handler._extract_model_from_cached_results([]) assert model_name is None @pytest.mark.asyncio async def test_async_responses_api_caching(): """ Test that responses API calls are properly cached and retrieved. This verifies the full cache lifecycle for ResponsesAPIResponse objects. """ # Setup cache setup_cache() caching_handler = LLMCachingHandler( original_function=aresponses, request_kwargs={}, start_time=datetime.now() ) # Create a mock ResponsesAPIResponse original_model = "gpt-4o" responses_api_response = ResponsesAPIResponse( id="resp_test123", created_at=int(time.time()), status="completed", model=original_model, object="response", output=[ { "type": "message", "id": "msg_123", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "This is a test response from the responses API.", "annotations": [], } ], } ], ) # Mock logging object logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aresponses.value, model=original_model, messages=[], # Responses API uses input, not messages function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) # Test parameters kwargs = { "model": original_model, "input": "Tell me a short story", "max_output_tokens": 100, "caching": True, } # Step 1: Cache the responses API response await caching_handler.async_set_cache( result=responses_api_response, original_function=aresponses, kwargs=kwargs ) await asyncio.sleep(0.5) # Step 2: Retrieve from cache cached_response = await caching_handler._async_get_cache( model=original_model, original_function=aresponses, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.aresponses.value, kwargs=kwargs, ) # Step 3: Verify the response is properly cached and retrieved assert cached_response.cached_result is not None assert isinstance(cached_response.cached_result, ResponsesAPIResponse) assert cached_response.cached_result.id == responses_api_response.id assert cached_response.cached_result.model == original_model assert cached_response.cached_result.status == "completed" assert len(cached_response.cached_result.output) == 1 # Verify cache hit flag is set assert cached_response.cached_result._hidden_params["cache_hit"] == True @pytest.mark.asyncio async def test_async_get_cache_updates_request_kwargs_for_streaming_responses(): """ Ensure streamed responses retain the normalized lookup kwargs so a later cache write can reuse the exact cache key from the read path. """ setup_cache() caching_handler = LLMCachingHandler( original_function=aresponses, request_kwargs={"stale": True}, start_time=datetime.now(), ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aresponses.value, model="gpt-4o", messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) kwargs = { "model": "gpt-4o", "input": "hello", "stream": True, "caching": True, } await caching_handler._async_get_cache( model="gpt-4o", original_function=aresponses, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.aresponses.value, kwargs=kwargs, ) assert "stale" not in caching_handler.request_kwargs assert caching_handler.request_kwargs["model"] == "gpt-4o" assert caching_handler.request_kwargs["input"] == "hello" assert caching_handler.request_kwargs["stream"] is True assert caching_handler.request_kwargs["cache_key"] == litellm.cache.get_cache_key( **caching_handler.request_kwargs ) def test_sync_responses_api_caching(): """ Test that synchronous responses API calls are properly cached and retrieved. """ # Setup cache setup_cache() caching_handler = LLMCachingHandler( original_function=responses, request_kwargs={}, start_time=datetime.now() ) # Create a mock ResponsesAPIResponse original_model = "gpt-4o" responses_api_response = ResponsesAPIResponse( id="resp_sync_test456", created_at=int(time.time()), status="completed", model=original_model, object="response", output=[ { "type": "message", "id": "msg_456", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "Sync response test.", "annotations": [], } ], } ], ) # Mock logging object logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.responses.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) # Test parameters kwargs = { "model": original_model, "input": "Tell me another story", "max_output_tokens": 100, "caching": True, } # Step 1: Cache the responses API response caching_handler.sync_set_cache(result=responses_api_response, kwargs=kwargs) # Step 2: Retrieve from cache cached_response = caching_handler._sync_get_cache( model=original_model, original_function=responses, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.responses.value, kwargs=kwargs, ) # Step 3: Verify the response is properly cached and retrieved assert cached_response.cached_result is not None assert isinstance(cached_response.cached_result, ResponsesAPIResponse) assert cached_response.cached_result.id == responses_api_response.id assert cached_response.cached_result.model == original_model assert cached_response.cached_result.status == "completed" # Verify cache hit flag is set assert cached_response.cached_result._hidden_params["cache_hit"] == True def test_convert_cached_responses_api_result_to_model_response(): """ Test that cached ResponsesAPIResponse results are properly converted back to ResponsesAPIResponse objects with correct structure. """ caching_handler = LLMCachingHandler( original_function=responses, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.responses.value, model="gpt-4o", messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) # Simulate cached result as a dictionary cached_result = { "id": "resp_convert_test789", "created_at": int(time.time()), "status": "completed", "model": "gpt-4o", "object": "response", "output": [ { "type": "message", "id": "msg_789", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "Conversion test response.", "annotations": [], } ], } ], } # Convert cached result to ResponsesAPIResponse result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=CallTypes.responses.value, kwargs={"model": "gpt-4o", "input": "test"}, logging_obj=logging_obj, model="gpt-4o", args=(), ) # Verify conversion assert isinstance(result, ResponsesAPIResponse) assert result.id == "resp_convert_test789" assert result.model == "gpt-4o" assert result.status == "completed" assert len(result.output) == 1 def test_sync_get_cache_does_not_eagerly_log_streaming_responses_hits(): litellm.set_verbose = True setup_cache() caching_handler = LLMCachingHandler( original_function=responses, request_kwargs={}, start_time=datetime.now() ) original_model = "gpt-4o" responses_api_response = ResponsesAPIResponse( id="resp_stream_sync_hit", created_at=int(time.time()), status="completed", model=original_model, object="response", output=[ { "type": "message", "id": "msg_stream_sync_hit", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "Sync streamed cache hit response.", "annotations": [], } ], } ], ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.responses.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) logging_obj.handle_sync_success_callbacks_for_async_calls = MagicMock() kwargs = { "model": original_model, "input": "Tell me a cached story", "stream": True, "caching": True, } caching_handler.sync_set_cache(result=responses_api_response, kwargs=kwargs) cached_response = caching_handler._sync_get_cache( model=original_model, original_function=responses, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.responses.value, kwargs=kwargs, ) assert cached_response.cached_result is not None assert isinstance( cached_response.cached_result, CachedResponsesAPIStreamingIterator ) logging_obj.handle_sync_success_callbacks_for_async_calls.assert_not_called() def test_sync_get_cache_defers_streaming_completion_hit_callbacks(): litellm.set_verbose = True setup_cache() caching_handler = LLMCachingHandler( original_function=completion, request_kwargs={}, start_time=datetime.now() ) original_model = "gpt-4o" logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.completion.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) logging_obj.handle_sync_success_callbacks_for_async_calls = MagicMock() kwargs = { "model": original_model, "messages": [{"role": "user", "content": "Tell me a cached joke"}], "stream": True, "caching": True, } caching_handler.sync_set_cache(result=chat_completion_response, kwargs=kwargs) cached_response = caching_handler._sync_get_cache( model=original_model, original_function=completion, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.completion.value, kwargs=kwargs, ) assert cached_response.cached_result is not None logging_obj.handle_sync_success_callbacks_for_async_calls.assert_not_called() def test_should_defer_streaming_cache_hit_callbacks_for_any_streaming_request(): assert ( _should_defer_streaming_cache_hit_callbacks( kwargs={"stream": True}, ) is True ) assert ( _should_defer_streaming_cache_hit_callbacks( kwargs={"stream": False}, ) is False ) assert ( _should_defer_streaming_cache_hit_callbacks( kwargs={}, ) is False ) @pytest.mark.asyncio async def test_async_get_cache_defers_streaming_completion_hit_callbacks(): litellm.set_verbose = True setup_cache() caching_handler = LLMCachingHandler( original_function=completion, request_kwargs={}, start_time=datetime.now() ) original_model = "gpt-4o" kwargs = { "model": original_model, "messages": [{"role": "user", "content": "Tell me a cached joke"}], "stream": True, "caching": True, } await caching_handler.async_set_cache( result=chat_completion_response, original_function=litellm.acompletion, kwargs=kwargs, ) await asyncio.sleep(0.2) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.acompletion.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) caching_handler._async_log_cache_hit_on_callbacks = MagicMock() cached_response = await caching_handler._async_get_cache( model=original_model, original_function=litellm.acompletion, logging_obj=logging_obj, start_time=datetime.now(), call_type=CallTypes.acompletion.value, kwargs=kwargs, ) assert cached_response is not None assert cached_response.cached_result is not None caching_handler._async_log_cache_hit_on_callbacks.assert_not_called() def test_convert_cached_streaming_responses_result_to_iterator(): """ Test that cached streaming Responses results are replayed through a synthetic streaming iterator instead of being returned as a full response object. """ caching_handler = LLMCachingHandler( original_function=responses, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.responses.value, model="gpt-4o", messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) cached_result = { "id": "resp_stream_cache_test", "created_at": int(time.time()), "status": "completed", "model": "gpt-4o", "object": "response", "output": [ { "type": "message", "id": "msg_stream_cache_test", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "Streaming cache replay test.", "annotations": [], } ], } ], } result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=CallTypes.responses.value, kwargs={"model": "gpt-4o", "input": "test", "stream": True}, logging_obj=logging_obj, model="gpt-4o", args=(), ) assert isinstance(result, CachedResponsesAPIStreamingIterator) assert result.completed_response is not None assert result.completed_response.response.id == cached_result["id"] streamed_events = list(result) assert streamed_events[0].type == "response.created" assert streamed_events[1].type == "response.in_progress" assert streamed_events[2].type == "response.output_item.added" assert streamed_events[3].type == "response.content_part.added" assert streamed_events[-4].type == "response.output_text.done" assert streamed_events[-3].type == "response.content_part.done" assert streamed_events[-2].type == "response.output_item.done" assert streamed_events[-1].type == "response.completed" assert streamed_events[-1].response.id == cached_result["id"] assert streamed_events[-1].response.output[0].content[0].text == ( "Streaming cache replay test." ) def test_is_chat_completion_cached_dict(): assert _is_chat_completion_cached_dict( {"id": "chatcmpl-abc", "object": "chat.completion", "choices": []} ) assert _is_chat_completion_cached_dict( {"id": "other", "object": "chat.completion.chunk", "choices": []} ) assert not _is_chat_completion_cached_dict( {"id": "resp_abc", "object": "response", "output": []} ) def test_convert_cached_aresponses_bridge_chat_completion_stream(): """ openai/responses chat-completions bridge caches ModelResponse JSON on aresponses cache keys; replay must not call ResponsesAPIResponse(**chatcmpl_dict). """ caching_handler = LLMCachingHandler( original_function=aresponses, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aresponses.value, model="gpt-5.4", messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) cached_result = { "id": "chatcmpl-bridge-cache-test", "object": "chat.completion", "created": int(time.time()), "model": "gpt-5.4", "choices": [ { "index": 0, "message": {"role": "assistant", "content": "Hi!"}, "finish_reason": "stop", } ], "usage": { "prompt_tokens": 7, "completion_tokens": 11, "total_tokens": 18, }, } result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=CallTypes.aresponses.value, kwargs={ "model": "gpt-5.4", "stream": True, "messages": [{"role": "user", "content": "hi"}], }, logging_obj=logging_obj, model="gpt-5.4", args=(), ) assert isinstance(result, CustomStreamWrapper) def test_convert_cached_streaming_reasoning_result_to_iterator(): caching_handler = LLMCachingHandler( original_function=responses, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.responses.value, model="gpt-4o", messages=[], function_id=str(uuid.uuid4()), stream=True, start_time=datetime.now(), ) cached_result = { "id": "resp_stream_reasoning_cache_test", "created_at": int(time.time()), "status": "completed", "model": "gpt-4o", "object": "response", "output": [ { "type": "reasoning", "id": "rs_stream_cache_test", "summary": [ { "type": "summary_text", "text": "Cached reasoning summary.", } ], } ], } result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=CallTypes.responses.value, kwargs={"model": "gpt-4o", "input": "test", "stream": True}, logging_obj=logging_obj, model="gpt-4o", args=(), ) assert isinstance(result, CachedResponsesAPIStreamingIterator) streamed_events = list(result) streamed_event_types = [ event.type.value if hasattr(event.type, "value") else str(event.type) for event in streamed_events ] assert streamed_event_types[:3] == [ "response.created", "response.in_progress", "response.output_item.added", ] assert streamed_event_types[-4:] == [ "response.reasoning_summary_text.done", "response.reasoning_summary_part.done", "response.output_item.done", "response.completed", ] assert streamed_event_types.count("response.reasoning_summary_text.delta") >= 1 delta_events = [ event for event in streamed_events if (event.type.value if hasattr(event.type, "value") else str(event.type)) == "response.reasoning_summary_text.delta" ] text_done_event = streamed_events[-4] part_done_event = streamed_events[-3] output_item_done_event = streamed_events[-2] assert all(delta_event.summary_index == 0 for delta_event in delta_events) assert text_done_event.text == "Cached reasoning summary." assert text_done_event.summary_index == 0 assert part_done_event.part.type == "summary_text" assert part_done_event.part.text == "Cached reasoning summary." assert output_item_done_event.item.type == "reasoning" assert output_item_done_event.item.summary[0]["text"] == "Cached reasoning summary." @pytest.mark.asyncio async def test_responses_api_cache_with_different_inputs(): """ Test that different inputs to the responses API result in different cache keys. This ensures cache isolation between different requests. """ # Setup cache setup_cache() caching_handler = LLMCachingHandler( original_function=aresponses, request_kwargs={}, start_time=datetime.now() ) original_model = "gpt-4o" # First request response_1 = ResponsesAPIResponse( id="resp_1", created_at=int(time.time()), status="completed", model=original_model, object="response", output=[ { "type": "message", "id": "msg_1", "status": "completed", "role": "assistant", "content": [ {"type": "output_text", "text": "Response 1", "annotations": []} ], } ], ) kwargs_1 = {"model": original_model, "input": "First unique input", "caching": True} await caching_handler.async_set_cache( result=response_1, original_function=aresponses, kwargs=kwargs_1 ) # Second request with different input response_2 = ResponsesAPIResponse( id="resp_2", created_at=int(time.time()), status="completed", model=original_model, object="response", output=[ { "type": "message", "id": "msg_2", "status": "completed", "role": "assistant", "content": [ {"type": "output_text", "text": "Response 2", "annotations": []} ], } ], ) kwargs_2 = { "model": original_model, "input": "Second unique input", "caching": True, } await caching_handler.async_set_cache( result=response_2, original_function=aresponses, kwargs=kwargs_2 ) await asyncio.sleep(0.5) # Retrieve both from cache logging_obj_1 = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aresponses.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) logging_obj_2 = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=CallTypes.aresponses.value, model=original_model, messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) cached_1 = await caching_handler._async_get_cache( model=original_model, original_function=aresponses, logging_obj=logging_obj_1, start_time=datetime.now(), call_type=CallTypes.aresponses.value, kwargs=kwargs_1, ) cached_2 = await caching_handler._async_get_cache( model=original_model, original_function=aresponses, logging_obj=logging_obj_2, start_time=datetime.now(), call_type=CallTypes.aresponses.value, kwargs=kwargs_2, ) # Verify each input gets its own cached response assert cached_1.cached_result is not None assert cached_2.cached_result is not None assert cached_1.cached_result.id == "resp_1" assert cached_2.cached_result.id == "resp_2" # Access output content properly (could be dict or object) output_1 = cached_1.cached_result.output[0] if isinstance(output_1, dict): text_1 = output_1["content"][0]["text"] else: text_1 = ( output_1.content[0].text if hasattr(output_1.content[0], "text") else output_1.content[0]["text"] ) output_2 = cached_2.cached_result.output[0] if isinstance(output_2, dict): text_2 = output_2["content"][0]["text"] else: text_2 = ( output_2.content[0].text if hasattr(output_2.content[0], "text") else output_2.content[0]["text"] ) assert text_1 == "Response 1" assert text_2 == "Response 2" @pytest.mark.parametrize( "call_type, cached_result, expected_type", [ ( CallTypes.responses.value, { "id": "resp_param_test", "created_at": 1234567890, "status": "completed", "model": "gpt-4o", "object": "response", "output": [ { "type": "message", "id": "msg_param", "status": "completed", "role": "assistant", "content": [ {"type": "output_text", "text": "Test", "annotations": []} ], } ], }, ResponsesAPIResponse, ), ( CallTypes.aresponses.value, { "id": "resp_async_param_test", "created_at": 1234567890, "status": "completed", "model": "gpt-4o", "object": "response", "output": [ { "type": "message", "id": "msg_async_param", "status": "completed", "role": "assistant", "content": [ { "type": "output_text", "text": "Async Test", "annotations": [], } ], } ], }, ResponsesAPIResponse, ), ], ) def test_convert_cached_responses_result_parameterized( call_type, cached_result, expected_type ): """ Parameterized test to verify both sync and async responses API cached results are converted to the correct ResponsesAPIResponse type. """ caching_handler = LLMCachingHandler( original_function=lambda: None, request_kwargs={}, start_time=datetime.now() ) logging_obj = LiteLLMLogging( litellm_call_id=str(datetime.now()), call_type=call_type, model="gpt-4o", messages=[], function_id=str(uuid.uuid4()), stream=False, start_time=datetime.now(), ) result = caching_handler._convert_cached_result_to_model_response( cached_result=cached_result, call_type=call_type, kwargs={}, logging_obj=logging_obj, model="gpt-4o", args=(), ) assert isinstance(result, expected_type) assert result is not None assert result.id == cached_result["id"] assert result.status == cached_result["status"]