6313512919

Committed 26 Sep 2023 01:37PM UTC coverage: 49.72% (-0.2%) from 49.967%

Build # 6313512919

Build Type

push

github

Committed by

web-flow

Commit Message

Add Pipelines async run (#5864)

* Add Pipeline.arun()

* Sleeper node

* Fix async running

* Add e2e tests

To run a Pipeline that doesn't have any async node in async mode:

    pytest e2e/pipelines/test_standard_pipelines.py::test_query_and_indexing_pipeline

To run a Pipeline that has a single async node in concurrent mode:

    pytest e2e/pipelines/test_standard_pipelines.py::test_async_concurrent_complex_pipeline

To run a Pipeline that has a single async node in sequential mode:

    pytest e2e/pipelines/test_standard_pipelines.py::test_async_sequential_complex_pipeline

* Remove unused _adispatch_run method

* Make Pipeline.run work with async nodes

* Revert "Make Pipeline.run work with async nodes"

This reverts commit 22d7a94e4.

* Rename Pipeline.arun to Pipeline._arun

* Enhance docstring

* Add Sleeper docstring

* Add release notes

* ignore typing across the node

* make pylint happy

* skip pylint on needed unused import

* fix

* if a node has an arun method, use it

---------

Co-authored-by: Massimiliano Pippi <mpippi@gmail.com>

Run Details

12242 of 24622 relevant lines covered (49.72%)

2.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

67.61

haystack/nodes/base.py

from __future__ import annotations
from typing import Any, Optional, Dict, List, Tuple, Union, Callable, Type

from copy import deepcopy
from abc import ABC, abstractmethod
from functools import wraps
import inspect
import logging

from haystack.schema import Document, MultiLabel
from haystack.errors import PipelineSchemaError
from haystack.utils.reflection import args_to_kwargs


logger = logging.getLogger(__name__)


def exportable_to_yaml(init_func):
    """
    Decorator that saves the init parameters of a node that later can
    be used with exporting YAML configuration of a Pipeline. We ensure
    that only params passed to the __init__ function of the implementation
    are saved, ignoring calls to the ancestors.
    """

    @wraps(init_func)
    def wrapper_exportable_to_yaml(self, *args, **kwargs):
        # Create the configuration dictionary if it doesn't exist yet
        if not self._component_config:
            self._component_config = {"params": {}, "type": type(self).__name__}

        # NOTE: inner classes constructor's __qualname__ will include the outer class' name,
        # e.g. "OuterClass.InnerClass.__init__". We then take only the last two parts of the
        # fully qualified name, in the previous example that would be "InnerClass.__init__"
        name_components = init_func.__qualname__.split(".")
        # Reconstruct the inner class' __qualname__ and compare with the __qualname__ of the implementation class.
        # If the number of components is wrong, let the IndexError bubble up, there's nothing we can do anyways.
        if f"{name_components[-2]}.{name_components[-1]}" == f"{self.__class__.__name__}.{init_func.__name__}":
            # Store all the input parameters in self._component_config
            args_as_kwargs = args_to_kwargs(args, init_func)
            params = {**args_as_kwargs, **kwargs}
            for k, v in params.items():
                self._component_config["params"][k] = v

        # Call the actual __init__ function with all the arguments
        init_func(self, *args, **kwargs)

    return wrapper_exportable_to_yaml


class BaseComponent(ABC):
    """
    A base class for implementing nodes in a Pipeline.
    """

    outgoing_edges: int
    _subclasses: dict = {}
    _component_config: dict = {}

    def __init__(self):
        component_params = self._component_config.get("params", {})
        self.outgoing_edges = self._calculate_outgoing_edges(component_params=component_params)

    # __init_subclass__ is invoked when a subclass of BaseComponent is _imported_
    # (not instantiated). It works approximately as a metaclass.
    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)

        # Each component must specify the number of outgoing edges (= different outputs).
        # During pipeline validation this number is compared to the requested number of output edges.
        if not hasattr(cls, "outgoing_edges"):
            raise ValueError(
                "BaseComponent subclasses must define the outgoing_edges class attribute. "
                "If this number depends on the component's parameters, make sure to override the _calculate_outgoing_edges() method. "
                "See https://haystack.deepset.ai/pipeline_nodes/custom-nodes for more information."
            )

        # Automatically registers all the init parameters in
        # an instance attribute called `_component_config`,
        # used to save this component to YAML. See exportable_to_yaml()
        cls.__init__ = exportable_to_yaml(cls.__init__)

        # Keeps track of all available subclasses by name.
        # Enables generic load() for all specific component implementations.
        # Registers abstract classes and base classes too.
        cls._subclasses[cls.__name__] = cls

    @property
    def name(self) -> Optional[str]:
        return self._component_config.get("name", None)

    @name.setter
    def name(self, value: str):
        self._component_config["name"] = value

    @property
    def utilized_components(self) -> List[BaseComponent]:
        if "params" not in self._component_config:
            return []
        return [param for param in self._component_config["params"].values() if isinstance(param, BaseComponent)]

    @property
    def type(self) -> str:
        return self._component_config["type"]

    def get_params(self, return_defaults: bool = False) -> Dict[str, Any]:
        component_signature = dict(inspect.signature(self.__class__).parameters)
        params: Dict[str, Any] = {}
        for key, value in self._component_config["params"].items():
            if value != component_signature[key].default or return_defaults:
                params[key] = value
        if return_defaults:
            for key, param in component_signature.items():
                if key not in params:
                    params[key] = param.default
        return params

    @classmethod
    def get_subclass(cls, component_type: str) -> Type[BaseComponent]:
        if component_type not in cls._subclasses.keys():
            raise PipelineSchemaError(
                f"Haystack component with the name '{component_type}' not found. "
                "Check the class name of your component for spelling mistakes and make sure you installed "
                "Haystack with the proper extras: https://docs.haystack.deepset.ai/docs/installation#custom-installation"
            )

        subclass = cls._subclasses[component_type]
        return subclass

    @classmethod
    def _calculate_outgoing_edges(cls, component_params: Dict[str, Any]) -> int:
        """
        Returns the number of outgoing edges for an instance of the component class given its component params.

        In some cases (e.g. RouteDocuments) the number of outgoing edges is not static but rather depends on its component params.
        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.
        Hence, this method is responsible for calculating the number of outgoing edges
        - during pipeline validation
        - to set the effective instance value of `outgoing_edges`.

        Override this method if the number of outgoing edges depends on the component params.
        If not overridden, returns the number of outgoing edges as defined in the component class.

        :param component_params: parameters to pass to the __init__() of the component.
        """
        return cls.outgoing_edges

    @classmethod
    def _create_instance(cls, component_type: str, component_params: Dict[str, Any], name: Optional[str] = None):
        """
        Returns an instance of the given subclass of BaseComponent.

        :param component_type: name of the component class to load.
        :param component_params: parameters to pass to the __init__() of the component.
        :param name: name of the component instance
        """
        subclass = cls.get_subclass(component_type)
        instance = subclass(**component_params)
        instance.name = name
        return instance

    @abstractmethod
    def run(
        self,
        query: Optional[str] = None,
        file_paths: Optional[List[str]] = None,
        labels: Optional[MultiLabel] = None,
        documents: Optional[List[Document]] = None,
        meta: Optional[dict] = None,
    ) -> Tuple[Dict, str]:
        """
        Method that will be executed when the node in the graph is called.

        The argument that are passed can vary between different types of nodes
        (e.g. retriever nodes expect different args than a reader node)


        See an example for an implementation in haystack/reader/base/BaseReader.py
        :return:
        """
        pass

    @abstractmethod
    def run_batch(
        self,
        queries: Optional[Union[str, List[str]]] = None,
        file_paths: Optional[List[str]] = None,
        labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,
        documents: Optional[Union[List[Document], List[List[Document]]]] = None,
        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
        params: Optional[dict] = None,
        debug: Optional[bool] = None,
    ):
        pass

    def _dispatch_run(self, **kwargs) -> Tuple[Dict, str]:
        """
        The Pipelines call this method when run() is executed. This method in turn executes the _dispatch_run_general()
        method with the correct run method.
        """
        return self._dispatch_run_general(self.run, **kwargs)

    def _dispatch_run_batch(self, **kwargs):
        """
        The Pipelines call this method when run_batch() is executed. This method in turn executes the
        _dispatch_run_general() method with the correct run method.
        """
        return self._dispatch_run_general(self.run_batch, **kwargs)

    def _dispatch_run_general(self, run_method: Callable, **kwargs):
        """
        This method takes care of the following:
          - inspect run_method's signature to validate if all necessary arguments are available
          - pop `debug` and sets them on the instance to control debug output
          - call run_method with the corresponding arguments and gather output
          - collate `_debug` information if present
          - merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline
        """
        arguments = deepcopy(kwargs)
        params = arguments.get("params") or {}

        run_signature_args = inspect.signature(run_method).parameters.keys()

        run_params: Dict[str, Any] = {}
        for key, value in params.items():
            if key == self.name:  # targeted params for this node
                if isinstance(value, dict):
                    # Extract debug attributes
                    if "debug" in value.keys():
                        self.debug = value.pop("debug")

                    for _k in value.keys():
                        if _k not in run_signature_args:
                            raise Exception(f"Invalid parameter '{_k}' for the node '{self.name}'.")

                run_params.update(**value)
            elif key in run_signature_args:  # global params
                run_params[key] = value

        run_inputs = {}
        for key, value in arguments.items():
            if key in run_signature_args:
                run_inputs[key] = value

        output, stream = run_method(**run_inputs, **run_params)

        # Collect debug information
        debug_info = {}
        if getattr(self, "debug", None):
            # Include input
            debug_info["input"] = {**run_inputs, **run_params}
            debug_info["input"]["debug"] = self.debug
            # Include output, exclude _debug to avoid recursion
            filtered_output = {key: value for key, value in output.items() if key != "_debug"}
            debug_info["output"] = filtered_output
        # Include custom debug info
        custom_debug = output.get("_debug", {})
        if custom_debug:
            debug_info["runtime"] = custom_debug

        # append _debug information from nodes
        all_debug = arguments.get("_debug", {})
        if debug_info:
            all_debug[self.name] = debug_info
        if all_debug:
            output["_debug"] = all_debug

        # 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() and k != "inputs":
                output[k] = v

        output["params"] = params
        return output, stream

    async def _adispatch_run_general(self, run_method: Callable, **kwargs):
        """
        This is the async version of _dispatch_run_general and is used indirectly by Pipeline._arun().
        When actually running the node it tries to run it asynchronously, if that fails fall back to a synchronous run.
        This makes it possible to run a pipeline asynchronously with a mix of async and sync nodes.

        This method takes care of the following:
          - inspect run_method's signature to validate if all necessary arguments are available
          - pop `debug` and sets them on the instance to control debug output
          - call run_method with the corresponding arguments and gather output
          - collate `_debug` information if present
          - merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline

        """
        arguments = deepcopy(kwargs)
        params = arguments.get("params") or {}

        run_signature_args = inspect.signature(run_method).parameters.keys()

        run_params: Dict[str, Any] = {}
        for key, value in params.items():
            if key == self.name:  # targeted params for this node
                if isinstance(value, dict):
                    # Extract debug attributes
                    if "debug" in value.keys():
                        self.debug = value.pop("debug")

                    for key in value.keys():
                        if key not in run_signature_args:
                            raise Exception(f"Invalid parameter '{key}' for the node '{self.name}'.")

                run_params.update(**value)
            elif key in run_signature_args:  # global params
                run_params[key] = value

        run_inputs = {}
        for key, value in arguments.items():
            if key in run_signature_args:
                run_inputs[key] = value

        try:
            output, stream = await run_method(**run_inputs, **run_params)
        except TypeError:
            output, stream = run_method(**run_inputs, **run_params)

        # Collect debug information
        debug_info = {}
        if getattr(self, "debug", None):
            # Include input
            debug_info["input"] = {**run_inputs, **run_params}
            debug_info["input"]["debug"] = self.debug
            # Include output, exclude _debug to avoid recursion
            filtered_output = {key: value for key, value in output.items() if key != "_debug"}
            debug_info["output"] = filtered_output
        # Include custom debug info
        custom_debug = output.get("_debug", {})
        if custom_debug:
            debug_info["runtime"] = custom_debug

        # append _debug information from nodes
        all_debug = arguments.get("_debug", {})
        if debug_info:
            all_debug[self.name] = debug_info
        if all_debug:
            output["_debug"] = all_debug

        # 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() and k != "inputs":
                output[k] = v

        output["params"] = params
        return output, stream


class RootNode(BaseComponent):
    """
    RootNode feeds inputs together with corresponding params to a Pipeline.
    """

    outgoing_edges = 1

    def run(self):  # type: ignore
        return {}, "output_1"

    def run_batch(self):  # type: ignore
        return {}, "output_1"

1	from __future__ import annotations	11✔
2	from typing import Any, Optional, Dict, List, Tuple, Union, Callable, Type	11✔
3
4	from copy import deepcopy	11✔
5	from abc import ABC, abstractmethod	11✔
6	from functools import wraps	11✔
7	import inspect	11✔
8	import logging	11✔
9
10	from haystack.schema import Document, MultiLabel	11✔
11	from haystack.errors import PipelineSchemaError	11✔
12	from haystack.utils.reflection import args_to_kwargs	11✔
13
14
15	logger = logging.getLogger(__name__)	11✔
16
17
18	def exportable_to_yaml(init_func):	11✔
19	"""
20	Decorator that saves the init parameters of a node that later can
21	be used with exporting YAML configuration of a Pipeline. We ensure
22	that only params passed to the __init__ function of the implementation
23	are saved, ignoring calls to the ancestors.
24	"""
25
26	@wraps(init_func)	11✔
27	def wrapper_exportable_to_yaml(self, args, *kwargs):	11✔
28	# Create the configuration dictionary if it doesn't exist yet
29	if not self._component_config:	11✔
30	self._component_config = {"params": {}, "type": type(self).__name__}	11✔
31
32	# NOTE: inner classes constructor's __qualname__ will include the outer class' name,
33	# e.g. "OuterClass.InnerClass.__init__". We then take only the last two parts of the
34	# fully qualified name, in the previous example that would be "InnerClass.__init__"
35	name_components = init_func.__qualname__.split(".")	11✔
36	# Reconstruct the inner class' __qualname__ and compare with the __qualname__ of the implementation class.
37	# If the number of components is wrong, let the IndexError bubble up, there's nothing we can do anyways.
38	if f"{name_components[-2]}.{name_components[-1]}" == f"{self.__class__.__name__}.{init_func.__name__}":	11✔
39	# Store all the input parameters in self._component_config
40	args_as_kwargs = args_to_kwargs(args, init_func)	11✔
41	params = {args_as_kwargs, kwargs}	11✔
42	for k, v in params.items():	11✔
43	self._component_config["params"][k] = v	11✔
44
45	# Call the actual __init__ function with all the arguments
46	init_func(self, args, *kwargs)	11✔
47
48	return wrapper_exportable_to_yaml	11✔
49
50
51	class BaseComponent(ABC):	11✔
52	"""
53	A base class for implementing nodes in a Pipeline.
54	"""
55
56	outgoing_edges: int	11✔
57	_subclasses: dict = {}	11✔
58	_component_config: dict = {}	11✔
59
60	def __init__(self):	11✔
61	component_params = self._component_config.get("params", {})	11✔
62	self.outgoing_edges = self._calculate_outgoing_edges(component_params=component_params)	11✔
63
64	# __init_subclass__ is invoked when a subclass of BaseComponent is _imported_
65	# (not instantiated). It works approximately as a metaclass.
66	def __init_subclass__(cls, **kwargs):	11✔
67	super().__init_subclass__(**kwargs)	11✔
68
69	# Each component must specify the number of outgoing edges (= different outputs).
70	# During pipeline validation this number is compared to the requested number of output edges.
71	if not hasattr(cls, "outgoing_edges"):	11✔
72	raise ValueError(	×
73	"BaseComponent subclasses must define the outgoing_edges class attribute. "
74	"If this number depends on the component's parameters, make sure to override the _calculate_outgoing_edges() method. "
75	"See https://haystack.deepset.ai/pipeline_nodes/custom-nodes for more information."
76	)
77
78	# Automatically registers all the init parameters in
79	# an instance attribute called `_component_config`,
80	# used to save this component to YAML. See exportable_to_yaml()
81	cls.__init__ = exportable_to_yaml(cls.__init__)	11✔
82
83	# Keeps track of all available subclasses by name.
84	# Enables generic load() for all specific component implementations.
85	# Registers abstract classes and base classes too.
86	cls._subclasses[cls.__name__] = cls	11✔
87
88	@property	11✔
89	def name(self) -> Optional[str]:	11✔
90	return self._component_config.get("name", None)	4✔
91
92	@name.setter	11✔
93	def name(self, value: str):	11✔
94	self._component_config["name"] = value	4✔
95
96	@property	11✔
97	def utilized_components(self) -> List[BaseComponent]:	11✔
98	if "params" not in self._component_config:	4✔
99	return []	×
100	return [param for param in self._component_config["params"].values() if isinstance(param, BaseComponent)]	4✔
101
102	@property	11✔
103	def type(self) -> str:	11✔
104	return self._component_config["type"]	4✔
105
106	def get_params(self, return_defaults: bool = False) -> Dict[str, Any]:	11✔
107	component_signature = dict(inspect.signature(self.__class__).parameters)	4✔
108	params: Dict[str, Any] = {}	4✔
109	for key, value in self._component_config["params"].items():	4✔
110	if value != component_signature[key].default or return_defaults:	3✔
111	params[key] = value	3✔
112	if return_defaults:	4✔
113	for key, param in component_signature.items():	×
114	if key not in params:	×
115	params[key] = param.default	×
116	return params	4✔
117
118	@classmethod	11✔
119	def get_subclass(cls, component_type: str) -> Type[BaseComponent]:	11✔
120	if component_type not in cls._subclasses.keys():	4✔
121	raise PipelineSchemaError(	1✔
122	f"Haystack component with the name '{component_type}' not found. "
123	"Check the class name of your component for spelling mistakes and make sure you installed "
124	"Haystack with the proper extras: https://docs.haystack.deepset.ai/docs/installation#custom-installation"
125	)
126
127	subclass = cls._subclasses[component_type]	4✔
128	return subclass	4✔
129
130	@classmethod	11✔
131	def _calculate_outgoing_edges(cls, component_params: Dict[str, Any]) -> int:	11✔
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	11✔
147
148	@classmethod	11✔
149	def _create_instance(cls, component_type: str, component_params: Dict[str, Any], name: Optional[str] = None):	11✔
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)	3✔
158	instance = subclass(**component_params)	3✔
159	instance.name = name	3✔
160	return instance	3✔
161
162	@abstractmethod	11✔
163	def run(	11✔
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	11✔
184	def run_batch(	11✔
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]:	11✔
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)	2✔
202
203	def _dispatch_run_batch(self, **kwargs):	11✔
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):	11✔
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)	2✔
220	params = arguments.get("params") or {}	2✔
221
222	run_signature_args = inspect.signature(run_method).parameters.keys()	2✔
223
224	run_params: Dict[str, Any] = {}	2✔
225	for key, value in params.items():	2✔
226	if key == self.name: # targeted params for this node	2✔
227	if isinstance(value, dict):	2✔
228	# Extract debug attributes
229	if "debug" in value.keys():	2✔
230	self.debug = value.pop("debug")	×
231
232	for _k in value.keys():	2✔
233	if _k not in run_signature_args:	2✔
234	raise Exception(f"Invalid parameter '{_k}' for the node '{self.name}'.")	1✔
235
236	run_params.update(**value)	2✔
237	elif key in run_signature_args: # global params	2✔
238	run_params[key] = value	×
239
240	run_inputs = {}	2✔
241	for key, value in arguments.items():	2✔
242	if key in run_signature_args:	2✔
243	run_inputs[key] = value	2✔
244
245	output, stream = run_method(run_inputs, run_params)	2✔
246
247	# Collect debug information
248	debug_info = {}	2✔
249	if getattr(self, "debug", None):	2✔
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", {})	2✔
258	if custom_debug:	2✔
259	debug_info["runtime"] = custom_debug	1✔
260
261	# append _debug information from nodes
262	all_debug = arguments.get("_debug", {})	2✔
263	if debug_info:	2✔
264	all_debug[self.name] = debug_info	1✔
265	if all_debug:	2✔
266	output["_debug"] = all_debug	1✔
267
268	# add "extra" args that were not used by the node, but not the 'inputs' value
269	for k, v in arguments.items():	2✔
270	if k not in output.keys() and k != "inputs":	2✔
271	output[k] = v	2✔
272
273	output["params"] = params	2✔
274	return output, stream	2✔
275
276	async def _adispatch_run_general(self, run_method: Callable, **kwargs):	11✔
277	"""
278	This is the async version of _dispatch_run_general and is used indirectly by Pipeline._arun().
279	When actually running the node it tries to run it asynchronously, if that fails fall back to a synchronous run.
280	This makes it possible to run a pipeline asynchronously with a mix of async and sync nodes.
281
282	This method takes care of the following:
283	- inspect run_method's signature to validate if all necessary arguments are available
284	- pop `debug` and sets them on the instance to control debug output
285	- call run_method with the corresponding arguments and gather output
286	- collate `_debug` information if present
287	- merge component output with the preceding output and pass it on to the subsequent Component in the Pipeline
288
289	"""
290	arguments = deepcopy(kwargs)	×
291	params = arguments.get("params") or {}	×
292
293	run_signature_args = inspect.signature(run_method).parameters.keys()	×
294
295	run_params: Dict[str, Any] = {}	×
296	for key, value in params.items():	×
297	if key == self.name: # targeted params for this node	×
298	if isinstance(value, dict):	×
299	# Extract debug attributes
300	if "debug" in value.keys():	×
301	self.debug = value.pop("debug")	×
302
303	for key in value.keys():	×
304	if key not in run_signature_args:	×
305	raise Exception(f"Invalid parameter '{key}' for the node '{self.name}'.")	×
306
307	run_params.update(**value)	×
308	elif key in run_signature_args: # global params	×
309	run_params[key] = value	×
310
311	run_inputs = {}	×
312	for key, value in arguments.items():	×
313	if key in run_signature_args:	×
314	run_inputs[key] = value	×
315
316	try:	×
317	output, stream = await run_method(run_inputs, run_params)	×
318	except TypeError:	×
319	output, stream = run_method(run_inputs, run_params)	×
320
321	# Collect debug information
322	debug_info = {}	×
323	if getattr(self, "debug", None):	×
324	# Include input
325	debug_info["input"] = {run_inputs, run_params}	×
326	debug_info["input"]["debug"] = self.debug	×
327	# Include output, exclude _debug to avoid recursion
328	filtered_output = {key: value for key, value in output.items() if key != "_debug"}	×
329	debug_info["output"] = filtered_output	×
330	# Include custom debug info
331	custom_debug = output.get("_debug", {})	×
332	if custom_debug:	×
333	debug_info["runtime"] = custom_debug	×
334
335	# append _debug information from nodes
336	all_debug = arguments.get("_debug", {})	×
337	if debug_info:	×
338	all_debug[self.name] = debug_info	×
339	if all_debug:	×
340	output["_debug"] = all_debug	×
341
342	# add "extra" args that were not used by the node, but not the 'inputs' value
343	for k, v in arguments.items():	×
344	if k not in output.keys() and k != "inputs":	×
345	output[k] = v	×
346
347	output["params"] = params	×
348	return output, stream	×
349
350
351	class RootNode(BaseComponent):	11✔
352	"""
353	RootNode feeds inputs together with corresponding params to a Pipeline.
354	"""
355
356	outgoing_edges = 1	11✔
357
358	def run(self): # type: ignore	11✔
359	return {}, "output_1"	2✔
360
361	def run_batch(self): # type: ignore	11✔
362	return {}, "output_1"	×

deepset-ai / haystack / 6313512919

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous