fix(streaming_handler.py): emit deep copy of completed chunk

litellm/litellm_core_utils/logging_utils.py

@@ -78,6 +78,9 @@ def _assemble_complete_response_from_streaming_chunks(
         Union[ModelResponse, TextCompletionResponse]
     ] = None
 
+    if isinstance(result, ModelResponse):
+        return result
+
     if result.choices[0].finish_reason is not None:  # if it's the last chunk
         streaming_chunks.append(result)
         try:

litellm/litellm_core_utils/streaming_handler.py

@@ -1472,6 +1472,15 @@ class CustomStreamWrapper:
         """
         self.logging_loop = loop
 
+    def cache_streaming_response(self, processed_chunk, cache_hit: bool):
+        """
+        Caches the streaming response
+        """
+        if not cache_hit and self.logging_obj._llm_caching_handler is not None:
+            self.logging_obj._llm_caching_handler._sync_add_streaming_response_to_cache(
+                processed_chunk
+            )
+
     def run_success_logging_and_cache_storage(self, processed_chunk, cache_hit: bool):
         """
         Runs success logging in a thread and adds the response to the cache
@@ -1503,12 +1512,6 @@ class CustomStreamWrapper:
         ## SYNC LOGGING
         self.logging_obj.success_handler(processed_chunk, None, None, cache_hit)
 
-        ## Sync store in cache
-        if self.logging_obj._llm_caching_handler is not None:
-            self.logging_obj._llm_caching_handler._sync_add_streaming_response_to_cache(
-                processed_chunk
-            )
-
     def finish_reason_handler(self):
         model_response = self.model_response_creator()
         _finish_reason = self.received_finish_reason or self.intermittent_finish_reason
@@ -1603,9 +1606,15 @@ class CustomStreamWrapper:
                         "usage",
                         getattr(complete_streaming_response, "usage"),
                     )
+                    self.cache_streaming_response(
+                        processed_chunk=complete_streaming_response.model_copy(
+                            deep=True
+                        ),
+                        cache_hit=cache_hit,
+                    )
                     executor.submit(
                         self.logging_obj.success_handler,
-                        complete_streaming_response,
+                        complete_streaming_response.model_copy(deep=True),
                         None,
                         None,
                         cache_hit,

tests/local_testing/test_caching.py

@@ -2151,7 +2151,7 @@ def test_logging_turn_off_message_logging_streaming():
     mock_obj = Cache(type="local")
     litellm.cache = mock_obj
 
-    with patch.object(mock_obj, "add_cache", new=MagicMock()) as mock_client:
+    with patch.object(mock_obj, "add_cache") as mock_client:
         print(f"mock_obj.add_cache: {mock_obj.add_cache}")
 
         resp = litellm.completion(
@@ -2167,7 +2167,7 @@ def test_logging_turn_off_message_logging_streaming():
         time.sleep(1)
 
         mock_client.assert_called_once()
-
+        print(f"mock_client.call_args: {mock_client.call_args}")
         assert mock_client.call_args.args[0].choices[0].message.content == "hello"