fix: Update deepcopying in Pipeline to have a fallback in case of error (#9346)

* First pass at fix for deepcopying inputs and outputs

* Add reno

* Add recursion for dict objects

* Bump recursion depth

* More tests and some improvments

* Fix unit tests

* PR comments

Author: sjrl

haystack/core/pipeline/async_pipeline.py

@@ -3,7 +3,6 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import asyncio
-from copy import deepcopy
 from typing import Any, AsyncIterator, Dict, List, Optional, Set
 
 from haystack import logging, tracing
@@ -16,6 +15,7 @@
     ComponentPriority,
     PipelineBase,
 )
+from haystack.core.pipeline.utils import deepcopy_with_fallback
 from haystack.telemetry import pipeline_running
 
 logger = logging.getLogger(__name__)
@@ -58,7 +58,7 @@ async def _run_component_async(  # pylint: disable=too-many-positional-arguments
         with PipelineBase._create_component_span(
             component_name=component_name, instance=instance, inputs=component_inputs, parent_span=parent_span
         ) as span:
-            span.set_content_tag(_COMPONENT_INPUT, deepcopy(component_inputs))
+            span.set_content_tag(_COMPONENT_INPUT, deepcopy_with_fallback(component_inputs))
             logger.info("Running component {component_name}", component_name=component_name)
 
             if getattr(instance, "__haystack_supports_async__", False):
@@ -76,7 +76,7 @@ async def _run_component_async(  # pylint: disable=too-many-positional-arguments
                 raise PipelineRuntimeError.from_invalid_output(component_name, instance.__class__, outputs)
 
             span.set_tag(_COMPONENT_VISITS, component_visits[component_name])
-            span.set_content_tag(_COMPONENT_OUTPUT, deepcopy(outputs))
+            span.set_content_tag(_COMPONENT_OUTPUT, deepcopy_with_fallback(outputs))
 
             return outputs
 
@@ -238,7 +238,7 @@ async def _run_highest_in_isolation(component_name: str) -> AsyncIterator[Dict[s
                         partial_result = finished.result()
                         scheduled_components.discard(finished_component_name)
                         if partial_result:
-                            yield_dict = {finished_component_name: deepcopy(partial_result)}
+                            yield_dict = {finished_component_name: deepcopy_with_fallback(partial_result)}
                             yield yield_dict  # partial outputs
 
                 if component_name in scheduled_components:
@@ -274,7 +274,7 @@ async def _run_highest_in_isolation(component_name: str) -> AsyncIterator[Dict[s
 
                 scheduled_components.remove(component_name)
                 if pruned:
-                    yield {component_name: deepcopy(pruned)}
+                    yield {component_name: deepcopy_with_fallback(pruned)}
 
             async def _schedule_task(component_name: str) -> None:
                 """
@@ -337,7 +337,7 @@ async def _wait_for_one_task_to_complete() -> AsyncIterator[Dict[str, Any]]:
                         partial_result = finished.result()
                         scheduled_components.discard(finished_component_name)
                         if partial_result:
-                            yield {finished_component_name: deepcopy(partial_result)}
+                            yield {finished_component_name: deepcopy_with_fallback(partial_result)}
 
             async def _wait_for_all_tasks_to_complete() -> AsyncIterator[Dict[str, Any]]:
                 """
@@ -350,7 +350,7 @@ async def _wait_for_all_tasks_to_complete() -> AsyncIterator[Dict[str, Any]]:
                         partial_result = finished.result()
                         scheduled_components.discard(finished_component_name)
                         if partial_result:
-                            yield {finished_component_name: deepcopy(partial_result)}
+                            yield {finished_component_name: deepcopy_with_fallback(partial_result)}
 
             # -------------------------------------------------
             # MAIN SCHEDULING LOOP
@@ -428,7 +428,7 @@ async def _wait_for_all_tasks_to_complete() -> AsyncIterator[Dict[str, Any]]:
                 yield partial_res
 
             # 4) Yield final pipeline outputs
-            yield deepcopy(pipeline_outputs)
+            yield deepcopy_with_fallback(pipeline_outputs)
 
     async def run_async(
         self, data: Dict[str, Any], include_outputs_from: Optional[Set[str]] = None, concurrency_limit: int = 4

haystack/core/pipeline/base.py

@@ -4,7 +4,6 @@
 
 import itertools
 from collections import defaultdict
-from copy import deepcopy
 from datetime import datetime
 from enum import IntEnum
 from pathlib import Path
@@ -33,7 +32,7 @@
     is_any_greedy_socket_ready,
     is_socket_lazy_variadic,
 )
-from haystack.core.pipeline.utils import FIFOPriorityQueue, parse_connect_string
+from haystack.core.pipeline.utils import FIFOPriorityQueue, deepcopy_with_fallback, parse_connect_string
 from haystack.core.serialization import DeserializationCallbacks, component_from_dict, component_to_dict
 from haystack.core.type_utils import _type_name, _types_are_compatible
 from haystack.marshal import Marshaller, YamlMarshaller
@@ -176,7 +175,7 @@ def from_dict(
         :returns:
             Deserialized component.
         """
-        data_copy = deepcopy(data)  # to prevent modification of original data
+        data_copy = deepcopy_with_fallback(data)  # to prevent modification of original data
         metadata = data_copy.get("metadata", {})
         max_runs_per_component = data_copy.get("max_runs_per_component", 100)
         connection_type_validation = data_copy.get("connection_type_validation", True)
@@ -895,7 +894,7 @@ def _prepare_component_input_data(self, data: Dict[str, Any]) -> Dict[str, Dict[
         # deepcopying the inputs prevents the Pipeline run logic from being altered unexpectedly
         # when the same input reference is passed to multiple components.
         for component_name, component_inputs in data.items():
-            data[component_name] = {k: deepcopy(v) for k, v in component_inputs.items()}
+            data[component_name] = {k: deepcopy_with_fallback(v) for k, v in component_inputs.items()}
 
         return data
 

haystack/core/pipeline/pipeline.py

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from copy import deepcopy
 from typing import Any, Dict, Mapping, Optional, Set, cast
 
 from haystack import logging, tracing
@@ -15,6 +14,7 @@
     ComponentPriority,
     PipelineBase,
 )
+from haystack.core.pipeline.utils import deepcopy_with_fallback
 from haystack.telemetry import pipeline_running
 
 logger = logging.getLogger(__name__)
@@ -54,7 +54,7 @@ def _run_component(
         ) as span:
             # We deepcopy the inputs otherwise we might lose that information
             # when we delete them in case they're sent to other Components
-            span.set_content_tag(_COMPONENT_INPUT, deepcopy(inputs))
+            span.set_content_tag(_COMPONENT_INPUT, deepcopy_with_fallback(inputs))
             logger.info("Running component {component_name}", component_name=component_name)
             try:
                 component_output = instance.run(**inputs)
@@ -252,7 +252,7 @@ def run(  # noqa: PLR0915, PLR0912
                 )
 
                 if component_pipeline_outputs:
-                    pipeline_outputs[component_name] = deepcopy(component_pipeline_outputs)
+                    pipeline_outputs[component_name] = deepcopy_with_fallback(component_pipeline_outputs)
                 if self._is_queue_stale(priority_queue):
                     priority_queue = self._fill_queue(ordered_component_names, inputs, component_visits)
 

haystack/core/pipeline/utils.py

@@ -3,9 +3,57 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import heapq
+from copy import deepcopy
 from itertools import count
 from typing import Any, List, Optional, Tuple
 
+from haystack import logging
+
+logger = logging.getLogger(__name__)
+
+
+def deepcopy_with_fallback(obj: Any, max_depth: Optional[int] = 100) -> Any:
+    """
+    Attempts to create a deep copy of the given object with a safe fallback mechanism.
+
+    If the object is a dictionary, each value is copied individually using this function recursively.
+    If an individual item fails to be copied, the original value is used as a fallback.
+    A maximum recursion depth is enforced to avoid deep or cyclic structures from causing issues.
+
+    :param obj: The object to attempt to deep copy.
+    :param max_depth: The maximum depth to recurse during deep copying. If 0, returns the object as-is.
+    :return: A deep copy of the object if successful; otherwise, the original object.
+    """
+    if max_depth is not None and max_depth <= 0:
+        return obj
+
+    try:
+        return deepcopy(obj)
+    except Exception as e:
+        # If the deepcopy fails we try to deepcopy each individual value if the object is a dictionary
+        next_depth = None if max_depth is None else max_depth - 1
+        if isinstance(obj, dict):
+            logger.info(
+                "Deepcopy failed for object of type '{obj_type}'. Error: {error}. Attempting item-wise copy.",
+                obj_type=type(obj).__name__,
+                error=e,
+            )
+            return {key: deepcopy_with_fallback(value, next_depth) for key, value in obj.items()}
+        elif isinstance(obj, (list, tuple, set)):
+            logger.info(
+                "Deepcopy failed for object of type '{obj_type}'. Error: {error}. Attempting item-wise copy.",
+                obj_type=type(obj).__name__,
+                error=e,
+            )
+            return type(obj)(deepcopy_with_fallback(item, next_depth) for item in obj)
+
+        logger.info(
+            "Deepcopy failed for object of type '{obj_type}'. Error: {error}. Returning original object instead.",
+            obj_type=type(obj).__name__,
+            error=e,
+        )
+        return obj
+
 
 def parse_connect_string(connection: str) -> Tuple[str, Optional[str]]:
     """

haystack/tools/component_tool.py

@@ -2,7 +2,6 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from copy import copy, deepcopy
 from dataclasses import fields, is_dataclass
 from inspect import getdoc
 from typing import Any, Callable, Dict, Optional, Union, get_args, get_origin
@@ -419,29 +418,3 @@ def _create_property_schema(self, python_type: Any, description: str, default: A
             schema["default"] = default
 
         return schema
-
-    def __deepcopy__(self, memo: Dict[Any, Any]) -> "ComponentTool":
-        # Jinja2 templates throw an Exception when we deepcopy them (see https://github.com/pallets/jinja/issues/758)
-        # When we use a ComponentTool in a pipeline at runtime, we deepcopy the tool
-        # We overwrite ComponentTool.__deepcopy__ to fix this until a more comprehensive fix is merged.
-        # We track the issue here: https://github.com/deepset-ai/haystack/issues/9011
-        result = copy(self)
-
-        # Add the object to the memo dictionary to handle circular references
-        memo[id(self)] = result
-
-        # Deep copy all attributes with exception handling
-        for key, value in self.__dict__.items():
-            try:
-                # Try to deep copy the attribute
-                setattr(result, key, deepcopy(value, memo))
-            except (TypeError, NotImplementedError):
-                # Fall back to using the original attribute for components that use Jinja2-templates
-                logger.debug(
-                    "deepcopy of ComponentTool {tool_name} failed. Using original attribute '{attribute}' instead.",
-                    tool_name=self.name,
-                    attribute=key,
-                )
-                setattr(result, key, getattr(self, key))
-
-        return result

releasenotes/notes/fix-deepcopy-in-pipeline-run-2b575c7b860f642f.yaml

@@ -0,0 +1,6 @@
+---
+fixes:
+  - |
+    Introduced `deepcopy_with_fallback` to improve robustness in the `Pipeline.run` method.
+    Previously, standard `deepcopy` was used on inputs and outputs to avoid unintended side effects from shared mutable data structures, as these values are accessed in multiple parts of the pipeline. However, certain Python objects cannot be deepcopied, which could lead to runtime errors.
+    The new `deepcopy_with_fallback` function attempts to deep copy an object and safely falls back to the original object if copying fails. This ensures pipeline execution remains stable while maintaining the protective behavior against shared-state modification.

test/core/pipeline/test_utils.py

@@ -2,9 +2,16 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
+import logging
 import pytest
 
-from haystack.core.pipeline.utils import parse_connect_string, FIFOPriorityQueue
+from haystack.components.builders.prompt_builder import PromptBuilder
+from haystack.core.pipeline.utils import parse_connect_string, FIFOPriorityQueue, deepcopy_with_fallback
+from haystack.tools import ComponentTool, Tool
+
+
+def get_weather_report(city: str) -> str:
+    return f"Weather report for {city}: 20°C, sunny"
 
 
 def test_parse_connection():
@@ -175,3 +182,80 @@ def test_queue_ordering_parametrized(empty_queue, items):
     sorted_items = sorted(items, key=lambda x: (x[0], items.index(x)))
     for priority, item in sorted_items:
         assert empty_queue.pop() == (priority, item)
+
+
+class TestDeepcopyWithFallback:
+    def test_deepcopy_with_fallback_copyable(self, caplog):
+        tool = Tool(
+            name="weather",
+            description="Get weather report",
+            parameters={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+            function=get_weather_report,
+        )
+        original = {"tools": tool}
+        with caplog.at_level(logging.INFO):
+            copy = deepcopy_with_fallback(original)
+            assert "Deepcopy failed for object of type" not in caplog.text
+        assert copy["tools"] == original["tools"]
+        # This should be a true copy so changing the name in the copy should not affect the original
+        copy["tools"].name = "copied_tool"
+        assert copy["tools"] != original["tools"]
+
+    def test_deepcopy_with_fallback_not_copyable(self, caplog):
+        problematic_tool = ComponentTool(
+            name="problematic_tool", description="This is a problematic tool.", component=PromptBuilder("{{query}}")
+        )
+        original = {"tools": problematic_tool}
+        with caplog.at_level(logging.INFO):
+            copy = deepcopy_with_fallback(original)
+            assert "Deepcopy failed for object of type 'ComponentTool'" in caplog.text
+        assert copy["tools"] == original["tools"]
+        # Not a true copy but a shallow copy so changing the name in the copy should also affect the original
+        copy["tools"].name = "copied_tool"
+        assert copy["tools"] == original["tools"]
+
+    def test_deepcopy_with_fallback_mixed_copyable_list(self, caplog):
+        tool1 = Tool(
+            name="tool1",
+            description="Get weather report",
+            parameters={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+            function=get_weather_report,
+        )
+        tool2 = ComponentTool(
+            name="problematic_tool", description="This is a problematic tool.", component=PromptBuilder("{{query}}")
+        )
+        original = {"tools": [tool1, tool2]}
+        with caplog.at_level(logging.INFO):
+            copy = deepcopy_with_fallback(original)
+            assert "Deepcopy failed for object of type 'ComponentTool'" in caplog.text
+        assert copy["tools"][0] == original["tools"][0]
+        # First tool should be a true copy so changing the name in the copy should not affect the original
+        copy["tools"][0].name = "copied_tool1"
+        assert copy["tools"][0] != original["tools"][0]
+        assert copy["tools"][1] == original["tools"][1]
+        # second should be a shallow copy so changing the name in the copy should also affect the original
+        copy["tools"][1].name = "copied_tool2"
+        assert copy["tools"][1] == original["tools"][1]
+
+    def test_deepcopy_with_fallback_mixed_copyable_tuple(self, caplog):
+        tool1 = Tool(
+            name="tool1",
+            description="Get weather report",
+            parameters={"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
+            function=get_weather_report,
+        )
+        tool2 = ComponentTool(
+            name="problematic_tool", description="This is a problematic tool.", component=PromptBuilder("{{query}}")
+        )
+        original = {"tools": (tool1, tool2)}
+        with caplog.at_level(logging.INFO):
+            copy = deepcopy_with_fallback(original)
+            assert "Deepcopy failed for object of type 'ComponentTool'" in caplog.text
+        assert copy["tools"][0] == original["tools"][0]
+        # First tool should be a true copy so changing the name in the copy should not affect the original
+        copy["tools"][0].name = "copied_tool1"
+        assert copy["tools"][0] != original["tools"][0]
+        assert copy["tools"][1] == original["tools"][1]
+        # second should be a shallow copy so changing the name in the copy should also affect the original
+        copy["tools"][1].name = "copied_tool2"
+        assert copy["tools"][1] == original["tools"][1]