We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Join Nodes produce incorrect results if preceded by another JoinNode
**Describe the bug**
We have a pipeline made up of four Retrievers, whose results are combined in two JoinDocuments nodes, then another JoinDocuments node to combine the results of those prior JoinDocuments.  Weights are specified for each Join node.

However, while the Retriever outputs are correct, the sequence of JoinNodes does not product all of the resulting documents.

We got around this by using just one JoinDocument node with all four inputs, but are reporting this bug because it could appear in other pipelines.

**Error message**
See example pipeline in https://github.com/JeffRisberg/HaystackPipelineTest.  There is no error message, however the results in the pipeline are missing documents.

**Expected behavior**
The resulting set of documents from any sequence of Joins should be the same as using JoinNode with all inputs.

**Additional context**
In the example pipeline at https://github.com/JeffRisberg/HaystackPipelineTest, I've simplified the problem down to using a simple InMemoryDocumentStore and a FilterRetriever, so that we have control over the documents.  The results are clearly wrong.

**To Reproduce**
Download https://github.com/JeffRisberg/HaystackPipelineTest and run main.py.  All 16 documents should be printed, but only 4 are.

**FAQ Check**
- [yes] Have you had a look at [our new FAQ page](https://haystack.deepset.ai/overview/faq)?

**System:**
 - OS: MacOS
 - GPU/CPU: CPU
 - Haystack version (commit or version number): 1.6.0
 - DocumentStore:  InMemoryDocumentStore
 - Reader: none
 - Retriever: FilterRetreiver


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `haystack/nodes/base.py`
Content:
```
1 from __future__ import annotations
2 from typing import Any, Optional, Dict, List, Tuple, Union, Callable, Type
3 
4 from copy import deepcopy
5 from abc import ABC, abstractmethod
6 from functools import wraps
7 import inspect
8 import logging
9 
10 from haystack.schema import Document, MultiLabel
11 from haystack.errors import PipelineSchemaError
12 from haystack.telemetry import send_custom_event
13 from haystack.utils import args_to_kwargs
14 
15 
16 logger = logging.getLogger(__name__)
17 
18 
19 def exportable_to_yaml(init_func):
20     """
21     Decorator that saves the init parameters of a node that later can
22     be used with exporting YAML configuration of a Pipeline. We ensure
23     that only params passed to the __init__ function of the implementation
24     are saved, ignoring calls to the ancestors.
25     """
26 
27     @wraps(init_func)
28     def wrapper_exportable_to_yaml(self, *args, **kwargs):
29 
30         # Create the configuration dictionary if it doesn't exist yet
31         if not self._component_config:
32             self._component_config = {"params": {}, "type": type(self).__name__}
33 
34         # NOTE: inner classes constructor's __qualname__ will include the outer class' name,
35         # e.g. "OuterClass.InnerClass.__init__". We then take only the last two parts of the
36         # fully qualified name, in the previous example that would be "InnerClass.__init__"
37         name_components = init_func.__qualname__.split(".")
38         # Reconstruct the inner class' __qualname__ and compare with the __qualname__ of the implementation class.
39         # If the number of components is wrong, let the IndexError bubble up, there's nothing we can do anyways.
40         if f"{name_components[-2]}.{name_components[-1]}" == f"{self.__class__.__name__}.{init_func.__name__}":
41             # Store all the input parameters in self._component_config
42             args_as_kwargs = args_to_kwargs(args, init_func)
43             params = {**args_as_kwargs, **kwargs}
44             for k, v in params.items():
45                 self._component_config["params"][k] = v
46 
47         # Call the actuall __init__ function with all the arguments
48         init_func(self, *args, **kwargs)
49 
50     return wrapper_exportable_to_yaml
51 
52 
53 class BaseComponent(ABC):
54     """
55     A base class for implementing nodes in a Pipeline.
56     """
57 
58     outgoing_edges: int
59     _subclasses: dict = {}
60     _component_config: dict = {}
61 
62     def __init__(self):
63         # a small subset of the component's parameters is sent in an event after applying filters defined in haystack.telemetry.NonPrivateParameters
64         component_params = self._component_config.get("params", {})
65         send_custom_event(event=f"{type(self).__name__} initialized", payload=component_params)
66         self.outgoing_edges = self._calculate_outgoing_edges(component_params=component_params)
67 
68     # __init_subclass__ is invoked when a subclass of BaseComponent is _imported_
69     # (not instantiated). It works approximately as a metaclass.
70     def __init_subclass__(cls, **kwargs):
71 
72         super().__init_subclass__(**kwargs)
73 
74         # Each component must specify the number of outgoing edges (= different outputs).
75         # During pipeline validation this number is compared to the requested number of output edges.
76         if not hasattr(cls, "outgoing_edges"):
77             raise ValueError(
78                 "BaseComponent subclasses must define the outgoing_edges class attribute. "
79                 "If this number depends on the component's parameters, make sure to override the _calculate_outgoing_edges() method. "
80                 "See https://haystack.deepset.ai/pipeline_nodes/custom-nodes for more information."
81             )
82 
83         # Automatically registers all the init parameters in
84         # an instance attribute called `_component_config`,
85         # used to save this component to YAML. See exportable_to_yaml()
86         cls.__init__ = exportable_to_yaml(cls.__init__)
87 
88         # Keeps track of all available subclasses by name.
89         # Enables generic load() for all specific component implementations.
90         # Registers abstract classes and base classes too.
91         cls._subclasses[cls.__name__] = cls
92 
93     @property
94     def name(self) -> Optional[str]:
95         return self._component_config.get("name", None)
96 
97     @name.setter
98     def name(self, value: str):
99         self._component_config["name"] = value
100 
101     @property
102     def utilized_components(self) -> List[BaseComponent]:
103         if "params" not in self._component_config:
104             return list()
105         return [param for param in self._component_config["params"].values() if isinstance(param, BaseComponent)]
106 
107     @property
108     def type(self) -> str:
109         return self._component_config["type"]
110 
111     def get_params(self, return_defaults: bool = False) -> Dict[str, Any]:
112         component_signature = dict(inspect.signature(self.__class__).parameters)
113         params: Dict[str, Any] = {}
114         for key, value in self._component_config["params"].items():
115             if value != component_signature[key].default or return_defaults:
116                 params[key] = value
117         if return_defaults:
118             for key, param in component_signature.items():
119                 if key not in params:
120                     params[key] = param.default
121         return params
122 
123     @classmethod
124     def get_subclass(cls, component_type: str) -> Type[BaseComponent]:
125         if component_type not in cls._subclasses.keys():
126             raise PipelineSchemaError(f"Haystack component with the name '{component_type}' not found.")
127         subclass = cls._subclasses[component_type]
128         return subclass
129 
130     @classmethod
131     def _calculate_outgoing_edges(cls, component_params: Dict[str, Any]) -> int:
132         """
133         Returns the number of outgoing edges for an instance of the component class given its component params.
134 
135         In some cases (e.g. RouteDocuments) the number of outgoing edges is not static but rather depends on its component params.
136         Setting the number of outgoing edges inside the constructor would not be sufficient, since it is already required for validating the pipeline when there is no instance yet.
137         Hence, this method is responsible for calculating the number of outgoing edges
138         - during pipeline validation
139         - to set the effective instance value of `outgoing_edges`.
140 
141         Override this method if the number of outgoing edges depends on the component params.
142         If not overridden, returns the number of outgoing edges as defined in the component class.
143 
144         :param component_params: parameters to pass to the __init__() of the component.
145         """
146         return cls.outgoing_edges
147 
148     @classmethod
149     def _create_instance(cls, component_type: str, component_params: Dict[str, Any], name: Optional[str] = None):
150         """
151         Returns an instance of the given subclass of BaseComponent.
152 
153         :param component_type: name of the component class to load.
154         :param component_params: parameters to pass to the __init__() of the component.
155         :param name: name of the component instance
156         """
157         subclass = cls.get_subclass(component_type)
158         instance = subclass(**component_params)
159         instance.name = name
160         return instance
161 
162     @abstractmethod
163     def run(
164         self,
165         query: Optional[str] = None,
166         file_paths: Optional[List[str]] = None,
167         labels: Optional[MultiLabel] = None,
168         documents: Optional[List[Document]] = None,
169         meta: Optional[dict] = None,
170     ) -> Tuple[Dict, str]:
171         """
172         Method that will be executed when the node in the graph is called.
173 
174         The argument that are passed can vary between different types of nodes
175         (e.g. retriever nodes expect different args than a reader node)
176 
177 
178         See an example for an implementation in haystack/reader/base/BaseReader.py
179         :return:
180         """
181         pass
182 
183     @abstractmethod
184     def run_batch(
185         self,
186         queries: Optional[Union[str, List[str]]] = None,
187         file_paths: Optional[List[str]] = None,
188         labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,
189         documents: Optional[Union[List[Document], List[List[Document]]]] = None,
190         meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
191         params: Optional[dict] = None,
192         debug: Optional[bool] = None,
193     ):
194         pass
195 
196     def _dispatch_run(self, **kwargs) -> Tuple[Dict, str]:
197         """
198         The Pipelines call this method when run() is executed. This method in turn executes the _dispatch_run_general()
199         method with the correct run method.
200         """
201         return self._dispatch_run_general(self.run, **kwargs)
202 
203     def _dispatch_run_batch(self, **kwargs):
204         """
205         The Pipelines call this method when run_batch() is executed. This method in turn executes the
206         _dispatch_run_general() method with the correct run method.
207         """
208         return self._dispatch_run_general(self.run_batch, **kwargs)
209 
210     def _dispatch_run_general(self, run_method: Callable, **kwargs):
211         """
212         This method takes care of the following:
213           - inspect run_method's signature to validate if all necessary arguments are available
214           - pop `debug` and sets them on the instance to control debug output
215           - call run_method with the corresponding arguments and gather output
216           - collate `_debug` information if present
217           - merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline
218         """
219         arguments = deepcopy(kwargs)
220         params = arguments.get("params") or {}
221 
222         run_signature_args = inspect.signature(run_method).parameters.keys()
223 
224         run_params: Dict[str, Any] = {}
225         for key, value in params.items():
226             if key == self.name:  # targeted params for this node
227                 if isinstance(value, dict):
228                     # Extract debug attributes
229                     if "debug" in value.keys():
230                         self.debug = value.pop("debug")
231 
232                     for _k, _v in value.items():
233                         if _k not in run_signature_args:
234                             raise Exception(f"Invalid parameter '{_k}' for the node '{self.name}'.")
235 
236                 run_params.update(**value)
237             elif key in run_signature_args:  # global params
238                 run_params[key] = value
239 
240         run_inputs = {}
241         for key, value in arguments.items():
242             if key in run_signature_args:
243                 run_inputs[key] = value
244 
245         output, stream = run_method(**run_inputs, **run_params)
246 
247         # Collect debug information
248         debug_info = {}
249         if getattr(self, "debug", None):
250             # Include input
251             debug_info["input"] = {**run_inputs, **run_params}
252             debug_info["input"]["debug"] = self.debug
253             # Include output, exclude _debug to avoid recursion
254             filtered_output = {key: value for key, value in output.items() if key != "_debug"}
255             debug_info["output"] = filtered_output
256         # Include custom debug info
257         custom_debug = output.get("_debug", {})
258         if custom_debug:
259             debug_info["runtime"] = custom_debug
260 
261         # append _debug information from nodes
262         all_debug = arguments.get("_debug", {})
263         if debug_info:
264             all_debug[self.name] = debug_info
265         if all_debug:
266             output["_debug"] = all_debug
267 
268         # add "extra" args that were not used by the node
269         for k, v in arguments.items():
270             if k not in output.keys():
271                 output[k] = v
272 
273         output["params"] = params
274         return output, stream
275 
276 
277 class RootNode(BaseComponent):
278     """
279     RootNode feeds inputs together with corresponding params to a Pipeline.
280     """
281 
282     outgoing_edges = 1
283 
284     def run(self):  # type: ignore
285         return {}, "output_1"
286 
287     def run_batch(self):  # type: ignore
288         return {}, "output_1"
289
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/haystack/nodes/base.py b/haystack/nodes/base.py
--- a/haystack/nodes/base.py
+++ b/haystack/nodes/base.py
@@ -265,9 +265,9 @@
         if all_debug:
             output["_debug"] = all_debug
 
-        # add "extra" args that were not used by the node
+        # add "extra" args that were not used by the node, but not the 'inputs' value
         for k, v in arguments.items():
-            if k not in output.keys():
+            if k not in output.keys() and k != "inputs":
                 output[k] = v
 
         output["params"] = params

{"golden_diff": "diff --git a/haystack/nodes/base.py b/haystack/nodes/base.py\n--- a/haystack/nodes/base.py\n+++ b/haystack/nodes/base.py\n@@ -265,9 +265,9 @@\n         if all_debug:\n             output[\"_debug\"] = all_debug\n \n-        # add \"extra\" args that were not used by the node\n+        # add \"extra\" args that were not used by the node, but not the 'inputs' value\n         for k, v in arguments.items():\n-            if k not in output.keys():\n+            if k not in output.keys() and k != \"inputs\":\n                 output[k] = v\n \n         output[\"params\"] = params\n", "issue": "Join Nodes produce incorrect results if preceded by another JoinNode\n**Describe the bug**\r\nWe have a pipeline made up of four Retrievers, whose results are combined in two JoinDocuments nodes, then another JoinDocuments node to combine the results of those prior JoinDocuments.  Weights are specified for each Join node.\r\n\r\nHowever, while the Retriever outputs are correct, the sequence of JoinNodes does not product all of the resulting documents.\r\n\r\nWe got around this by using just one JoinDocument node with all four inputs, but are reporting this bug because it could appear in other pipelines.\r\n\r\n**Error message**\r\nSee example pipeline in https://github.com/JeffRisberg/HaystackPipelineTest.  There is no error message, however the results in the pipeline are missing documents.\r\n\r\n**Expected behavior**\r\nThe resulting set of documents from any sequence of Joins should be the same as using JoinNode with all inputs.\r\n\r\n**Additional context**\r\nIn the example pipeline at https://github.com/JeffRisberg/HaystackPipelineTest, I've simplified the problem down to using a simple InMemoryDocumentStore and a FilterRetriever, so that we have control over the documents.  The results are clearly wrong.\r\n\r\n**To Reproduce**\r\nDownload https://github.com/JeffRisberg/HaystackPipelineTest and run main.py.  All 16 documents should be printed, but only 4 are.\r\n\r\n**FAQ Check**\r\n- [yes] Have you had a look at [our new FAQ page](https://haystack.deepset.ai/overview/faq)?\r\n\r\n**System:**\r\n - OS: MacOS\r\n - GPU/CPU: CPU\r\n - Haystack version (commit or version number): 1.6.0\r\n - DocumentStore:  InMemoryDocumentStore\r\n - Reader: none\r\n - Retriever: FilterRetreiver\r\n\n", "before_files": [{"content": "from __future__ import annotations\nfrom typing import Any, Optional, Dict, List, Tuple, Union, Callable, Type\n\nfrom copy import deepcopy\nfrom abc import ABC, abstractmethod\nfrom functools import wraps\nimport inspect\nimport logging\n\nfrom haystack.schema import Document, MultiLabel\nfrom haystack.errors import PipelineSchemaError\nfrom haystack.telemetry import send_custom_event\nfrom haystack.utils import args_to_kwargs\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef exportable_to_yaml(init_func):\n    \"\"\"\n    Decorator that saves the init parameters of a node that later can\n    be used with exporting YAML configuration of a Pipeline. We ensure\n    that only params passed to the __init__ function of the implementation\n    are saved, ignoring calls to the ancestors.\n    \"\"\"\n\n    @wraps(init_func)\n    def wrapper_exportable_to_yaml(self, *args, **kwargs):\n\n        # Create the configuration dictionary if it doesn't exist yet\n        if not self._component_config:\n            self._component_config = {\"params\": {}, \"type\": type(self).__name__}\n\n        # NOTE: inner classes constructor's __qualname__ will include the outer class' name,\n        # e.g. \"OuterClass.InnerClass.__init__\". We then take only the last two parts of the\n        # fully qualified name, in the previous example that would be \"InnerClass.__init__\"\n        name_components = init_func.__qualname__.split(\".\")\n        # Reconstruct the inner class' __qualname__ and compare with the __qualname__ of the implementation class.\n        # If the number of components is wrong, let the IndexError bubble up, there's nothing we can do anyways.\n        if f\"{name_components[-2]}.{name_components[-1]}\" == f\"{self.__class__.__name__}.{init_func.__name__}\":\n            # Store all the input parameters in self._component_config\n            args_as_kwargs = args_to_kwargs(args, init_func)\n            params = {**args_as_kwargs, **kwargs}\n            for k, v in params.items():\n                self._component_config[\"params\"][k] = v\n\n        # Call the actuall __init__ function with all the arguments\n        init_func(self, *args, **kwargs)\n\n    return wrapper_exportable_to_yaml\n\n\nclass BaseComponent(ABC):\n    \"\"\"\n    A base class for implementing nodes in a Pipeline.\n    \"\"\"\n\n    outgoing_edges: int\n    _subclasses: dict = {}\n    _component_config: dict = {}\n\n    def __init__(self):\n        # a small subset of the component's parameters is sent in an event after applying filters defined in haystack.telemetry.NonPrivateParameters\n        component_params = self._component_config.get(\"params\", {})\n        send_custom_event(event=f\"{type(self).__name__} initialized\", payload=component_params)\n        self.outgoing_edges = self._calculate_outgoing_edges(component_params=component_params)\n\n    # __init_subclass__ is invoked when a subclass of BaseComponent is _imported_\n    # (not instantiated). It works approximately as a metaclass.\n    def __init_subclass__(cls, **kwargs):\n\n        super().__init_subclass__(**kwargs)\n\n        # Each component must specify the number of outgoing edges (= different outputs).\n        # During pipeline validation this number is compared to the requested number of output edges.\n        if not hasattr(cls, \"outgoing_edges\"):\n            raise ValueError(\n                \"BaseComponent subclasses must define the outgoing_edges class attribute. \"\n                \"If this number depends on the component's parameters, make sure to override the _calculate_outgoing_edges() method. \"\n                \"See https://haystack.deepset.ai/pipeline_nodes/custom-nodes for more information.\"\n            )\n\n        # Automatically registers all the init parameters in\n        # an instance attribute called `_component_config`,\n        # used to save this component to YAML. See exportable_to_yaml()\n        cls.__init__ = exportable_to_yaml(cls.__init__)\n\n        # Keeps track of all available subclasses by name.\n        # Enables generic load() for all specific component implementations.\n        # Registers abstract classes and base classes too.\n        cls._subclasses[cls.__name__] = cls\n\n    @property\n    def name(self) -> Optional[str]:\n        return self._component_config.get(\"name\", None)\n\n    @name.setter\n    def name(self, value: str):\n        self._component_config[\"name\"] = value\n\n    @property\n    def utilized_components(self) -> List[BaseComponent]:\n        if \"params\" not in self._component_config:\n            return list()\n        return [param for param in self._component_config[\"params\"].values() if isinstance(param, BaseComponent)]\n\n    @property\n    def type(self) -> str:\n        return self._component_config[\"type\"]\n\n    def get_params(self, return_defaults: bool = False) -> Dict[str, Any]:\n        component_signature = dict(inspect.signature(self.__class__).parameters)\n        params: Dict[str, Any] = {}\n        for key, value in self._component_config[\"params\"].items():\n            if value != component_signature[key].default or return_defaults:\n                params[key] = value\n        if return_defaults:\n            for key, param in component_signature.items():\n                if key not in params:\n                    params[key] = param.default\n        return params\n\n    @classmethod\n    def get_subclass(cls, component_type: str) -> Type[BaseComponent]:\n        if component_type not in cls._subclasses.keys():\n            raise PipelineSchemaError(f\"Haystack component with the name '{component_type}' not found.\")\n        subclass = cls._subclasses[component_type]\n        return subclass\n\n    @classmethod\n    def _calculate_outgoing_edges(cls, component_params: Dict[str, Any]) -> int:\n        \"\"\"\n        Returns the number of outgoing edges for an instance of the component class given its component params.\n\n        In some cases (e.g. RouteDocuments) the number of outgoing edges is not static but rather depends on its component params.\n        Setting the number of outgoing edges inside the constructor would not be sufficient, since it is already required for validating the pipeline when there is no instance yet.\n        Hence, this method is responsible for calculating the number of outgoing edges\n        - during pipeline validation\n        - to set the effective instance value of `outgoing_edges`.\n\n        Override this method if the number of outgoing edges depends on the component params.\n        If not overridden, returns the number of outgoing edges as defined in the component class.\n\n        :param component_params: parameters to pass to the __init__() of the component.\n        \"\"\"\n        return cls.outgoing_edges\n\n    @classmethod\n    def _create_instance(cls, component_type: str, component_params: Dict[str, Any], name: Optional[str] = None):\n        \"\"\"\n        Returns an instance of the given subclass of BaseComponent.\n\n        :param component_type: name of the component class to load.\n        :param component_params: parameters to pass to the __init__() of the component.\n        :param name: name of the component instance\n        \"\"\"\n        subclass = cls.get_subclass(component_type)\n        instance = subclass(**component_params)\n        instance.name = name\n        return instance\n\n    @abstractmethod\n    def run(\n        self,\n        query: Optional[str] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[MultiLabel] = None,\n        documents: Optional[List[Document]] = None,\n        meta: Optional[dict] = None,\n    ) -> Tuple[Dict, str]:\n        \"\"\"\n        Method that will be executed when the node in the graph is called.\n\n        The argument that are passed can vary between different types of nodes\n        (e.g. retriever nodes expect different args than a reader node)\n\n\n        See an example for an implementation in haystack/reader/base/BaseReader.py\n        :return:\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def run_batch(\n        self,\n        queries: Optional[Union[str, List[str]]] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,\n        documents: Optional[Union[List[Document], List[List[Document]]]] = None,\n        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,\n        params: Optional[dict] = None,\n        debug: Optional[bool] = None,\n    ):\n        pass\n\n    def _dispatch_run(self, **kwargs) -> Tuple[Dict, str]:\n        \"\"\"\n        The Pipelines call this method when run() is executed. This method in turn executes the _dispatch_run_general()\n        method with the correct run method.\n        \"\"\"\n        return self._dispatch_run_general(self.run, **kwargs)\n\n    def _dispatch_run_batch(self, **kwargs):\n        \"\"\"\n        The Pipelines call this method when run_batch() is executed. This method in turn executes the\n        _dispatch_run_general() method with the correct run method.\n        \"\"\"\n        return self._dispatch_run_general(self.run_batch, **kwargs)\n\n    def _dispatch_run_general(self, run_method: Callable, **kwargs):\n        \"\"\"\n        This method takes care of the following:\n          - inspect run_method's signature to validate if all necessary arguments are available\n          - pop `debug` and sets them on the instance to control debug output\n          - call run_method with the corresponding arguments and gather output\n          - collate `_debug` information if present\n          - merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline\n        \"\"\"\n        arguments = deepcopy(kwargs)\n        params = arguments.get(\"params\") or {}\n\n        run_signature_args = inspect.signature(run_method).parameters.keys()\n\n        run_params: Dict[str, Any] = {}\n        for key, value in params.items():\n            if key == self.name:  # targeted params for this node\n                if isinstance(value, dict):\n                    # Extract debug attributes\n                    if \"debug\" in value.keys():\n                        self.debug = value.pop(\"debug\")\n\n                    for _k, _v in value.items():\n                        if _k not in run_signature_args:\n                            raise Exception(f\"Invalid parameter '{_k}' for the node '{self.name}'.\")\n\n                run_params.update(**value)\n            elif key in run_signature_args:  # global params\n                run_params[key] = value\n\n        run_inputs = {}\n        for key, value in arguments.items():\n            if key in run_signature_args:\n                run_inputs[key] = value\n\n        output, stream = run_method(**run_inputs, **run_params)\n\n        # Collect debug information\n        debug_info = {}\n        if getattr(self, \"debug\", None):\n            # Include input\n            debug_info[\"input\"] = {**run_inputs, **run_params}\n            debug_info[\"input\"][\"debug\"] = self.debug\n            # Include output, exclude _debug to avoid recursion\n            filtered_output = {key: value for key, value in output.items() if key != \"_debug\"}\n            debug_info[\"output\"] = filtered_output\n        # Include custom debug info\n        custom_debug = output.get(\"_debug\", {})\n        if custom_debug:\n            debug_info[\"runtime\"] = custom_debug\n\n        # append _debug information from nodes\n        all_debug = arguments.get(\"_debug\", {})\n        if debug_info:\n            all_debug[self.name] = debug_info\n        if all_debug:\n            output[\"_debug\"] = all_debug\n\n        # add \"extra\" args that were not used by the node\n        for k, v in arguments.items():\n            if k not in output.keys():\n                output[k] = v\n\n        output[\"params\"] = params\n        return output, stream\n\n\nclass RootNode(BaseComponent):\n    \"\"\"\n    RootNode feeds inputs together with corresponding params to a Pipeline.\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def run(self):  # type: ignore\n        return {}, \"output_1\"\n\n    def run_batch(self):  # type: ignore\n        return {}, \"output_1\"\n", "path": "haystack/nodes/base.py"}], "after_files": [{"content": "from __future__ import annotations\nfrom typing import Any, Optional, Dict, List, Tuple, Union, Callable, Type\n\nfrom copy import deepcopy\nfrom abc import ABC, abstractmethod\nfrom functools import wraps\nimport inspect\nimport logging\n\nfrom haystack.schema import Document, MultiLabel\nfrom haystack.errors import PipelineSchemaError\nfrom haystack.telemetry import send_custom_event\nfrom haystack.utils import args_to_kwargs\n\n\nlogger = logging.getLogger(__name__)\n\n\ndef exportable_to_yaml(init_func):\n    \"\"\"\n    Decorator that saves the init parameters of a node that later can\n    be used with exporting YAML configuration of a Pipeline. We ensure\n    that only params passed to the __init__ function of the implementation\n    are saved, ignoring calls to the ancestors.\n    \"\"\"\n\n    @wraps(init_func)\n    def wrapper_exportable_to_yaml(self, *args, **kwargs):\n\n        # Create the configuration dictionary if it doesn't exist yet\n        if not self._component_config:\n            self._component_config = {\"params\": {}, \"type\": type(self).__name__}\n\n        # NOTE: inner classes constructor's __qualname__ will include the outer class' name,\n        # e.g. \"OuterClass.InnerClass.__init__\". We then take only the last two parts of the\n        # fully qualified name, in the previous example that would be \"InnerClass.__init__\"\n        name_components = init_func.__qualname__.split(\".\")\n        # Reconstruct the inner class' __qualname__ and compare with the __qualname__ of the implementation class.\n        # If the number of components is wrong, let the IndexError bubble up, there's nothing we can do anyways.\n        if f\"{name_components[-2]}.{name_components[-1]}\" == f\"{self.__class__.__name__}.{init_func.__name__}\":\n            # Store all the input parameters in self._component_config\n            args_as_kwargs = args_to_kwargs(args, init_func)\n            params = {**args_as_kwargs, **kwargs}\n            for k, v in params.items():\n                self._component_config[\"params\"][k] = v\n\n        # Call the actuall __init__ function with all the arguments\n        init_func(self, *args, **kwargs)\n\n    return wrapper_exportable_to_yaml\n\n\nclass BaseComponent(ABC):\n    \"\"\"\n    A base class for implementing nodes in a Pipeline.\n    \"\"\"\n\n    outgoing_edges: int\n    _subclasses: dict = {}\n    _component_config: dict = {}\n\n    def __init__(self):\n        # a small subset of the component's parameters is sent in an event after applying filters defined in haystack.telemetry.NonPrivateParameters\n        component_params = self._component_config.get(\"params\", {})\n        send_custom_event(event=f\"{type(self).__name__} initialized\", payload=component_params)\n        self.outgoing_edges = self._calculate_outgoing_edges(component_params=component_params)\n\n    # __init_subclass__ is invoked when a subclass of BaseComponent is _imported_\n    # (not instantiated). It works approximately as a metaclass.\n    def __init_subclass__(cls, **kwargs):\n\n        super().__init_subclass__(**kwargs)\n\n        # Each component must specify the number of outgoing edges (= different outputs).\n        # During pipeline validation this number is compared to the requested number of output edges.\n        if not hasattr(cls, \"outgoing_edges\"):\n            raise ValueError(\n                \"BaseComponent subclasses must define the outgoing_edges class attribute. \"\n                \"If this number depends on the component's parameters, make sure to override the _calculate_outgoing_edges() method. \"\n                \"See https://haystack.deepset.ai/pipeline_nodes/custom-nodes for more information.\"\n            )\n\n        # Automatically registers all the init parameters in\n        # an instance attribute called `_component_config`,\n        # used to save this component to YAML. See exportable_to_yaml()\n        cls.__init__ = exportable_to_yaml(cls.__init__)\n\n        # Keeps track of all available subclasses by name.\n        # Enables generic load() for all specific component implementations.\n        # Registers abstract classes and base classes too.\n        cls._subclasses[cls.__name__] = cls\n\n    @property\n    def name(self) -> Optional[str]:\n        return self._component_config.get(\"name\", None)\n\n    @name.setter\n    def name(self, value: str):\n        self._component_config[\"name\"] = value\n\n    @property\n    def utilized_components(self) -> List[BaseComponent]:\n        if \"params\" not in self._component_config:\n            return list()\n        return [param for param in self._component_config[\"params\"].values() if isinstance(param, BaseComponent)]\n\n    @property\n    def type(self) -> str:\n        return self._component_config[\"type\"]\n\n    def get_params(self, return_defaults: bool = False) -> Dict[str, Any]:\n        component_signature = dict(inspect.signature(self.__class__).parameters)\n        params: Dict[str, Any] = {}\n        for key, value in self._component_config[\"params\"].items():\n            if value != component_signature[key].default or return_defaults:\n                params[key] = value\n        if return_defaults:\n            for key, param in component_signature.items():\n                if key not in params:\n                    params[key] = param.default\n        return params\n\n    @classmethod\n    def get_subclass(cls, component_type: str) -> Type[BaseComponent]:\n        if component_type not in cls._subclasses.keys():\n            raise PipelineSchemaError(f\"Haystack component with the name '{component_type}' not found.\")\n        subclass = cls._subclasses[component_type]\n        return subclass\n\n    @classmethod\n    def _calculate_outgoing_edges(cls, component_params: Dict[str, Any]) -> int:\n        \"\"\"\n        Returns the number of outgoing edges for an instance of the component class given its component params.\n\n        In some cases (e.g. RouteDocuments) the number of outgoing edges is not static but rather depends on its component params.\n        Setting the number of outgoing edges inside the constructor would not be sufficient, since it is already required for validating the pipeline when there is no instance yet.\n        Hence, this method is responsible for calculating the number of outgoing edges\n        - during pipeline validation\n        - to set the effective instance value of `outgoing_edges`.\n\n        Override this method if the number of outgoing edges depends on the component params.\n        If not overridden, returns the number of outgoing edges as defined in the component class.\n\n        :param component_params: parameters to pass to the __init__() of the component.\n        \"\"\"\n        return cls.outgoing_edges\n\n    @classmethod\n    def _create_instance(cls, component_type: str, component_params: Dict[str, Any], name: Optional[str] = None):\n        \"\"\"\n        Returns an instance of the given subclass of BaseComponent.\n\n        :param component_type: name of the component class to load.\n        :param component_params: parameters to pass to the __init__() of the component.\n        :param name: name of the component instance\n        \"\"\"\n        subclass = cls.get_subclass(component_type)\n        instance = subclass(**component_params)\n        instance.name = name\n        return instance\n\n    @abstractmethod\n    def run(\n        self,\n        query: Optional[str] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[MultiLabel] = None,\n        documents: Optional[List[Document]] = None,\n        meta: Optional[dict] = None,\n    ) -> Tuple[Dict, str]:\n        \"\"\"\n        Method that will be executed when the node in the graph is called.\n\n        The argument that are passed can vary between different types of nodes\n        (e.g. retriever nodes expect different args than a reader node)\n\n\n        See an example for an implementation in haystack/reader/base/BaseReader.py\n        :return:\n        \"\"\"\n        pass\n\n    @abstractmethod\n    def run_batch(\n        self,\n        queries: Optional[Union[str, List[str]]] = None,\n        file_paths: Optional[List[str]] = None,\n        labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,\n        documents: Optional[Union[List[Document], List[List[Document]]]] = None,\n        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,\n        params: Optional[dict] = None,\n        debug: Optional[bool] = None,\n    ):\n        pass\n\n    def _dispatch_run(self, **kwargs) -> Tuple[Dict, str]:\n        \"\"\"\n        The Pipelines call this method when run() is executed. This method in turn executes the _dispatch_run_general()\n        method with the correct run method.\n        \"\"\"\n        return self._dispatch_run_general(self.run, **kwargs)\n\n    def _dispatch_run_batch(self, **kwargs):\n        \"\"\"\n        The Pipelines call this method when run_batch() is executed. This method in turn executes the\n        _dispatch_run_general() method with the correct run method.\n        \"\"\"\n        return self._dispatch_run_general(self.run_batch, **kwargs)\n\n    def _dispatch_run_general(self, run_method: Callable, **kwargs):\n        \"\"\"\n        This method takes care of the following:\n          - inspect run_method's signature to validate if all necessary arguments are available\n          - pop `debug` and sets them on the instance to control debug output\n          - call run_method with the corresponding arguments and gather output\n          - collate `_debug` information if present\n          - merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline\n        \"\"\"\n        arguments = deepcopy(kwargs)\n        params = arguments.get(\"params\") or {}\n\n        run_signature_args = inspect.signature(run_method).parameters.keys()\n\n        run_params: Dict[str, Any] = {}\n        for key, value in params.items():\n            if key == self.name:  # targeted params for this node\n                if isinstance(value, dict):\n                    # Extract debug attributes\n                    if \"debug\" in value.keys():\n                        self.debug = value.pop(\"debug\")\n\n                    for _k, _v in value.items():\n                        if _k not in run_signature_args:\n                            raise Exception(f\"Invalid parameter '{_k}' for the node '{self.name}'.\")\n\n                run_params.update(**value)\n            elif key in run_signature_args:  # global params\n                run_params[key] = value\n\n        run_inputs = {}\n        for key, value in arguments.items():\n            if key in run_signature_args:\n                run_inputs[key] = value\n\n        output, stream = run_method(**run_inputs, **run_params)\n\n        # Collect debug information\n        debug_info = {}\n        if getattr(self, \"debug\", None):\n            # Include input\n            debug_info[\"input\"] = {**run_inputs, **run_params}\n            debug_info[\"input\"][\"debug\"] = self.debug\n            # Include output, exclude _debug to avoid recursion\n            filtered_output = {key: value for key, value in output.items() if key != \"_debug\"}\n            debug_info[\"output\"] = filtered_output\n        # Include custom debug info\n        custom_debug = output.get(\"_debug\", {})\n        if custom_debug:\n            debug_info[\"runtime\"] = custom_debug\n\n        # append _debug information from nodes\n        all_debug = arguments.get(\"_debug\", {})\n        if debug_info:\n            all_debug[self.name] = debug_info\n        if all_debug:\n            output[\"_debug\"] = all_debug\n\n        # add \"extra\" args that were not used by the node, but not the 'inputs' value\n        for k, v in arguments.items():\n            if k not in output.keys() and k != \"inputs\":\n                output[k] = v\n\n        output[\"params\"] = params\n        return output, stream\n\n\nclass RootNode(BaseComponent):\n    \"\"\"\n    RootNode feeds inputs together with corresponding params to a Pipeline.\n    \"\"\"\n\n    outgoing_edges = 1\n\n    def run(self):  # type: ignore\n        return {}, \"output_1\"\n\n    def run_batch(self):  # type: ignore\n        return {}, \"output_1\"\n", "path": "haystack/nodes/base.py"}]}