fix(responses): resolve deepcopy error with tool_choice ValidatorIterator (#17192) (#17205)

Replace copy.deepcopy with model_dump + model_validate in streaming
iterator logging to handle Pydantic ValidatorIterator objects that
cannot be pickled when tool_choice uses allowed_tools mode.

Co-authored-by: Krish Dholakia <krrishdholakia@gmail.com>

litellm/responses/streaming_iterator.py

@@ -382,11 +382,20 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
 
     def _handle_logging_completed_response(self):
         """Handle logging for completed responses in async context"""
-        # Create a deep copy for logging to avoid modifying the response object that will be returned to the user
+        # Create a copy for logging to avoid modifying the response object that will be returned to the user
         # The logging handlers may transform usage from Responses API format (input_tokens/output_tokens)
         # to chat completion format (prompt_tokens/completion_tokens) for internal logging
-        import copy
-        logging_response = copy.deepcopy(self.completed_response)
+        # Use model_dump + model_validate instead of deepcopy to avoid pickle errors with
+        # Pydantic ValidatorIterator when response contains tool_choice with allowed_tools (fixes #17192)
+        logging_response = self.completed_response
+        if self.completed_response is not None and hasattr(self.completed_response, 'model_dump'):
+            try:
+                logging_response = type(self.completed_response).model_validate(
+                    self.completed_response.model_dump()
+                )
+            except Exception:
+                # Fallback to original if serialization fails
+                pass
 
         asyncio.create_task(
             self.logging_obj.async_success_handler(
@@ -468,11 +477,20 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator):
 
     def _handle_logging_completed_response(self):
         """Handle logging for completed responses in sync context"""
-        # Create a deep copy for logging to avoid modifying the response object that will be returned to the user
+        # Create a copy for logging to avoid modifying the response object that will be returned to the user
         # The logging handlers may transform usage from Responses API format (input_tokens/output_tokens)
         # to chat completion format (prompt_tokens/completion_tokens) for internal logging
-        import copy
-        logging_response = copy.deepcopy(self.completed_response)
+        # Use model_dump + model_validate instead of deepcopy to avoid pickle errors with
+        # Pydantic ValidatorIterator when response contains tool_choice with allowed_tools (fixes #17192)
+        logging_response = self.completed_response
+        if self.completed_response is not None and hasattr(self.completed_response, 'model_dump'):
+            try:
+                logging_response = type(self.completed_response).model_validate(
+                    self.completed_response.model_dump()
+                )
+            except Exception:
+                # Fallback to original if serialization fails
+                pass
 
         run_async_function(
             async_function=self.logging_obj.async_success_handler,

tests/llm_responses_api_testing/test_base_responses_api_streaming_iterator.py

@@ -231,7 +231,7 @@ class TestBaseResponsesAPIStreamingIterator:
         mock_logging_obj = Mock(spec=LiteLLMLoggingObj)
         mock_logging_obj.model_call_details = {"litellm_params": {}}
         mock_config = Mock(spec=BaseResponsesAPIConfig)
-        
+
         # Create the iterator instance
         iterator = BaseResponsesAPIStreamingIterator(
             response=mock_response,
@@ -239,15 +239,78 @@ class TestBaseResponsesAPIStreamingIterator:
             responses_api_provider_config=mock_config,
             logging_obj=mock_logging_obj
         )
-        
+
         # Test with empty chunk
         result = iterator._process_chunk("")
         assert result is None
-        
+
         # Test with None chunk
         result = iterator._process_chunk(None)
         assert result is None
 
+    def test_handle_logging_completed_response_with_unpickleable_objects(self):
+        """
+        Test that _handle_logging_completed_response handles responses containing
+        objects that cannot be pickled (like Pydantic ValidatorIterator).
+
+        This test verifies the fix for issue #17192 where streaming with tool_choice
+        containing allowed_tools would fail with:
+        "cannot pickle 'pydantic_core._pydantic_core.ValidatorIterator' object"
+
+        The fix uses model_dump + model_validate instead of copy.deepcopy.
+        """
+        import asyncio
+        from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator
+
+        # Mock dependencies
+        mock_response = Mock()
+        mock_response.headers = {}
+        mock_response.aiter_lines = Mock()
+        mock_logging_obj = Mock(spec=LiteLLMLoggingObj)
+        mock_logging_obj.model_call_details = {"litellm_params": {}}
+        mock_logging_obj.async_success_handler = Mock()
+        mock_logging_obj.success_handler = Mock()
+        mock_config = Mock(spec=BaseResponsesAPIConfig)
+
+        # Create the iterator instance
+        iterator = ResponsesAPIStreamingIterator(
+            response=mock_response,
+            model="gpt-4",
+            responses_api_provider_config=mock_config,
+            logging_obj=mock_logging_obj,
+            litellm_metadata={"model_info": {"id": "model_123"}},
+            custom_llm_provider="openai"
+        )
+
+        # Create a ResponseCompletedEvent with tool_choice that has model_dump
+        mock_completed_response = Mock()
+        mock_completed_response.model_dump.return_value = {
+            "type": "response.completed",
+            "response": {
+                "id": "resp_123",
+                "output": [{"type": "function_call", "name": "search_web"}],
+                "tool_choice": {"type": "function", "name": "search_web"}
+            }
+        }
+        # model_validate should return a new mock (the copy)
+        type(mock_completed_response).model_validate = Mock(return_value=Mock())
+
+        iterator.completed_response = mock_completed_response
+
+        # This should NOT raise an exception
+        # Previously it would fail with: TypeError: cannot pickle 'ValidatorIterator'
+        # Mock asyncio.create_task and executor.submit since we're not in async context
+        with patch('asyncio.create_task') as mock_create_task, \
+             patch('litellm.responses.streaming_iterator.executor') as mock_executor:
+            try:
+                iterator._handle_logging_completed_response()
+            except TypeError as e:
+                if "pickle" in str(e):
+                    pytest.fail(f"_handle_logging_completed_response failed with pickle error: {e}")
+                raise
+
+        # Verify model_dump was called (our fix uses this instead of deepcopy)
+        mock_completed_response.model_dump.assert_called_once()
     def test_process_chunk_exception_does_not_call_handle_failure(self):
         """
         Test that _process_chunk raises exceptions without calling _handle_failure.
@@ -294,4 +357,4 @@ class TestBaseResponsesAPIStreamingIterator:
             
             # Verify _handle_failure was NOT called in _process_chunk
             # It should only be called by the outer exception handler in __next__/__anext__
-            mock_handle_failure.assert_not_called()
+            mock_handle_failure.assert_not_called()