Coverage for haystack/core/pipeline/component_checks.py: 100%
66 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-21 13:53 +0000
1# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2#
3# SPDX-License-Identifier: Apache-2.0
5from typing import Any
7from haystack.core.component.types import InputSocket
10class _NoOutputProduced:
11 """
12 Marker a socket input holds when its sender ran without producing a value for that socket.
14 Test for it with `isinstance(value, _NoOutputProduced)`, never with `==`: comparing runs `__eq__` on the value
15 being checked, which for a value such as a DataFrame does not return a bool. It carries no state, so instances
16 are interchangeable and every one of them means the same thing.
18 It is serializable because it reaches a pipeline snapshot as part of the pipeline's inputs.
19 """
21 def __eq__(self, other: object) -> bool:
22 return isinstance(other, _NoOutputProduced)
24 def __hash__(self) -> int:
25 return hash(_NoOutputProduced)
27 def to_dict(self) -> dict[str, Any]:
28 """
29 Serialize the marker. It carries no state, so the payload is empty.
30 """
31 return {}
33 @classmethod
34 def from_dict(cls, data: dict[str, Any]) -> "_NoOutputProduced": # noqa: ARG003
35 """
36 Deserialize the marker.
37 """
38 return cls()
41def can_component_run(component: dict, inputs: dict) -> bool:
42 """
43 Checks if the component can run, given the current state of its inputs.
45 A component needs to pass two gates so that it is ready to run:
46 1. It has received all mandatory inputs.
47 2. It has received a trigger.
48 :param component: Component metadata and the component instance.
49 :param inputs: Inputs for the component.
50 """
51 received_all_mandatory_inputs = are_all_sockets_ready(component, inputs, only_check_mandatory=True)
52 received_trigger = has_any_trigger(component, inputs)
54 return received_all_mandatory_inputs and received_trigger
57def has_any_trigger(component: dict, inputs: dict) -> bool:
58 """
59 Checks if a component was triggered to execute.
61 There are 3 triggers:
62 1. A predecessor provided input to the component.
63 2. Input to the component was provided from outside the pipeline (e.g. user input).
64 3. The component does not receive input from any other components in the pipeline and `Pipeline.run` was called.
66 A trigger can only cause a component to execute ONCE because:
67 1. Components consume inputs from predecessors before execution (they are deleted).
68 2. Inputs from outside the pipeline can only trigger a component when it is executed for the first time.
69 3. `Pipeline.run` can only trigger a component when it is executed for the first time.
71 :param component: Component metadata and the component instance.
72 :param inputs: Inputs for the component.
73 """
74 trigger_from_predecessor = any_predecessors_provided_input(component, inputs)
75 trigger_from_user = has_user_input(inputs) and component["visits"] == 0
76 trigger_without_inputs = can_not_receive_inputs_from_pipeline(component) and component["visits"] == 0
78 return trigger_from_predecessor or trigger_from_user or trigger_without_inputs
81def are_all_sockets_ready(component: dict, inputs: dict, only_check_mandatory: bool = False) -> bool:
82 """
83 Checks if all sockets of a component have enough inputs for the component to execute.
85 :param component: Component metadata and the component instance.
86 :param inputs: Inputs for the component.
87 :param only_check_mandatory: If only mandatory sockets should be checked.
88 """
89 filled_sockets = set()
90 expected_sockets = set()
91 if only_check_mandatory:
92 sockets_to_check = {
93 socket_name: socket for socket_name, socket in component["input_sockets"].items() if socket.is_mandatory
94 }
95 else:
96 sockets_to_check = {
97 socket_name: socket
98 for socket_name, socket in component["input_sockets"].items()
99 if socket.is_mandatory or len(socket.senders)
100 }
102 for socket_name, socket in sockets_to_check.items():
103 socket_inputs = inputs.get(socket_name, [])
104 expected_sockets.add(socket_name)
106 # Check if socket has all required inputs or is a lazy variadic socket with any input
107 if has_socket_received_all_inputs(socket, socket_inputs) or (
108 socket.is_lazy_variadic and any_socket_input_received(socket_inputs)
109 ):
110 filled_sockets.add(socket_name)
112 return filled_sockets == expected_sockets
115def any_predecessors_provided_input(component: dict, inputs: dict) -> bool:
116 """
117 Checks if a component received inputs from any predecessors.
119 :param component: Component metadata and the component instance.
120 :param inputs: Inputs for the component.
121 """
122 return any(
123 any_socket_value_from_predecessor_received(inputs.get(socket_name, []))
124 for socket_name in component["input_sockets"].keys()
125 )
128def any_socket_value_from_predecessor_received(socket_inputs: list[dict[str, Any]]) -> bool:
129 """
130 Checks if a component socket received input from any predecessors.
132 :param socket_inputs: Inputs for the component's socket.
133 """
134 # When sender is None, the input was provided from outside the pipeline.
135 return any(not isinstance(inp["value"], _NoOutputProduced) and inp["sender"] is not None for inp in socket_inputs)
138def has_user_input(inputs: dict) -> bool:
139 """
140 Checks if a component has received input from outside the pipeline (e.g. user input).
142 :param inputs: Inputs for the component.
143 """
144 return any(inp for socket in inputs.values() for inp in socket if inp["sender"] is None)
147def can_not_receive_inputs_from_pipeline(component: dict) -> bool:
148 """
149 Checks if a component can not receive inputs from any other components in the pipeline.
151 :param: Component metadata and the component instance.
152 """
153 return all(len(sock.senders) == 0 for sock in component["input_sockets"].values())
156def all_socket_predecessors_executed(socket: InputSocket, socket_inputs: list[dict[str, Any]]) -> bool:
157 """
158 Checks if all components connecting to an InputSocket have executed.
160 :param: The InputSocket of a component.
161 :param: socket_inputs: Inputs for the socket.
162 """
163 expected_senders = set(socket.senders)
164 executed_senders = {inp["sender"] for inp in socket_inputs if inp["sender"] is not None}
165 return expected_senders == executed_senders
168def any_socket_input_received(socket_inputs: list[dict]) -> bool:
169 """
170 Checks if a socket has received any input from any other components in the pipeline or from outside the pipeline.
172 :param socket_inputs: Inputs for the socket.
173 """
174 return any(not isinstance(inp["value"], _NoOutputProduced) for inp in socket_inputs)
177def has_lazy_variadic_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool:
178 """
179 Checks if a lazy variadic socket has received all expected inputs from other components in the pipeline.
181 :param socket: The InputSocket of a component.
182 :param socket_inputs: Inputs for the socket.
183 """
184 expected_senders = set(socket.senders)
185 actual_senders = {
186 sock["sender"]
187 for sock in socket_inputs
188 if not isinstance(sock["value"], _NoOutputProduced) and sock["sender"] is not None
189 }
190 return expected_senders == actual_senders
193def has_socket_received_all_inputs(socket: InputSocket, socket_inputs: list[dict]) -> bool:
194 """
195 Checks if a socket has received all expected inputs.
197 :param socket: The InputSocket of a component.
198 :param socket_inputs: Inputs for the socket.
199 """
200 # No inputs received for the socket, it is not filled.
201 if len(socket_inputs) == 0:
202 return False
204 # The socket is greedy variadic and at least one input was produced, it is complete.
205 if (
206 socket.is_variadic
207 and socket.is_greedy
208 and any(not isinstance(sock["value"], _NoOutputProduced) for sock in socket_inputs)
209 ):
210 return True
212 # The socket is lazy variadic and all expected inputs were produced.
213 if socket.is_lazy_variadic and has_lazy_variadic_socket_received_all_inputs(socket, socket_inputs):
214 return True
216 # The socket is not variadic and the only expected input is complete.
217 return not socket.is_variadic and not isinstance(socket_inputs[0]["value"], _NoOutputProduced)
220def all_predecessors_executed(component: dict, inputs: dict) -> bool:
221 """
222 Checks if all predecessors of a component have executed.
224 :param component: Component metadata and the component instance.
225 :param inputs: Inputs for the component.
226 """
227 return all(
228 all_socket_predecessors_executed(socket, inputs.get(socket_name, []))
229 for socket_name, socket in component["input_sockets"].items()
230 )
233def is_any_greedy_socket_ready(component: dict, inputs: dict) -> bool:
234 """
235 Checks if the component has any greedy socket that is ready to run.
237 :param component: Component metadata and the component instance.
238 :param inputs: Inputs for the component.
239 """
240 for socket_name, socket in component["input_sockets"].items():
241 if socket.is_greedy and has_socket_received_all_inputs(socket, inputs.get(socket_name, [])):
242 return True
244 return False