Coverage for haystack/core/pipeline/base.py: 92%
636 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
5import itertools
6import json
7from collections import defaultdict
8from collections.abc import Iterator, Mapping, Sequence
9from contextlib import AbstractContextManager as ContextManager
10from datetime import datetime
11from enum import IntEnum
12from pathlib import Path
13from typing import Any, TextIO, TypeVar, Union, get_args
15import networkx
16from typing_extensions import Self
18from haystack import logging, tracing
19from haystack.core.component import Component, InputSocket, OutputSocket, component
20from haystack.core.errors import (
21 DeserializationError,
22 PipelineComponentsBlockedError,
23 PipelineConnectError,
24 PipelineDrawingError,
25 PipelineError,
26 PipelineMaxComponentRuns,
27 PipelineRuntimeError,
28 PipelineValidationError,
29)
30from haystack.core.pipeline.component_checks import (
31 _NoOutputProduced,
32 all_predecessors_executed,
33 are_all_sockets_ready,
34 can_component_run,
35 is_any_greedy_socket_ready,
36)
37from haystack.core.pipeline.utils import FIFOPriorityQueue, _deepcopy_with_exceptions, parse_connect_string
38from haystack.core.serialization import (
39 DeserializationCallbacks,
40 component_from_dict,
41 component_to_dict,
42 generate_qualified_class_name,
43)
44from haystack.core.serialization_security import (
45 _check_module_allowed,
46 _deserialization_context,
47 mark_deserialization_internal,
48)
49from haystack.core.type_utils import (
50 ConversionStrategyType,
51 _convert_value,
52 _safe_get_origin,
53 _type_name,
54 _types_are_compatible,
55)
56from haystack.marshal import Marshaller, YamlMarshaller
57from haystack.utils import is_in_jupyter, type_serialization
59from .descriptions import find_pipeline_inputs, find_pipeline_outputs
60from .draw import _to_mermaid_image
62DEFAULT_MARSHALLER = YamlMarshaller()
64# We use a generic type to annotate the return value of class methods,
65# so that static analyzers won't be confused when derived classes
66# use those methods.
67T = TypeVar("T", bound="PipelineBase")
69logger = logging.getLogger(__name__)
72# Constants for tracing tags
73_COMPONENT_INPUT = "haystack.component.input"
74_COMPONENT_OUTPUT = "haystack.component.output"
75_COMPONENT_VISITS = "haystack.component.visits"
77InputsType = dict[str, dict[str, list[dict[str, Any]]]]
80class ComponentPriority(IntEnum):
81 HIGHEST = 1
82 READY = 2
83 DEFER = 3
84 BLOCKED = 4
87class PipelineBase: # noqa: PLW1641
88 """
89 Components orchestration engine.
91 Builds a graph of components and orchestrates their execution according to the execution graph.
92 """
94 def __init__(
95 self,
96 metadata: dict[str, Any] | None = None,
97 max_runs_per_component: int = 100,
98 connection_type_validation: bool = True,
99 ) -> None:
100 """
101 Creates the Pipeline.
103 :param metadata:
104 Arbitrary dictionary to store metadata about this `Pipeline`. Make sure all the values contained in
105 this dictionary can be serialized and deserialized if you wish to save this `Pipeline` to file.
106 :param max_runs_per_component:
107 How many times the `Pipeline` can run the same Component.
108 If this limit is reached a `PipelineMaxComponentRuns` exception is raised.
109 If not set defaults to 100 runs per Component.
110 :param connection_type_validation: Whether the pipeline will validate the types of the connections.
111 Defaults to True.
112 """
113 self._telemetry_runs = 0
114 self._last_telemetry_sent: datetime | None = None
115 self.metadata = metadata or {}
116 self.graph = networkx.MultiDiGraph()
117 self._max_runs_per_component = max_runs_per_component
118 self._connection_type_validation = connection_type_validation
120 def __eq__(self, other: object) -> bool:
121 """
122 Pipeline equality is defined by their type and the equality of their serialized form.
124 Pipelines of the same type share every metadata, node and edge, but they're not required to use
125 the same node instances: this allows pipeline saved and then loaded back to be equal to themselves.
126 """
127 if not isinstance(other, type(self)):
128 return False
129 return self.to_dict() == other.to_dict()
131 def __repr__(self) -> str:
132 """
133 Returns a text representation of the Pipeline.
134 """
135 res = f"{object.__repr__(self)}\n"
136 if self.metadata:
137 res += "🧱 Metadata\n"
138 for k, v in self.metadata.items():
139 res += f" - {k}: {v}\n"
141 res += "🚅 Components\n"
142 for name, instance in self.graph.nodes(data="instance"):
143 res += f" - {name}: {instance.__class__.__name__}\n"
145 res += "🛤️ Connections\n"
146 for sender, receiver, edge_data in self.graph.edges(data=True):
147 sender_socket = edge_data["from_socket"].name
148 receiver_socket = edge_data["to_socket"].name
149 res += f" - {sender}.{sender_socket} -> {receiver}.{receiver_socket} ({edge_data['conn_type']})\n"
151 return res
153 def to_dict(self) -> dict[str, Any]:
154 """
155 Serializes the pipeline to a dictionary.
157 This is meant to be an intermediate representation but it can be also used to save a pipeline to file.
159 :returns:
160 Dictionary with serialized data.
161 """
162 components = {}
163 for name, instance in self.graph.nodes(data="instance"):
164 components[name] = component_to_dict(instance, name)
166 connections = []
167 for sender, receiver, edge_data in self.graph.edges.data():
168 sender_socket = edge_data["from_socket"].name
169 receiver_socket = edge_data["to_socket"].name
170 connections.append({"sender": f"{sender}.{sender_socket}", "receiver": f"{receiver}.{receiver_socket}"})
171 return {
172 "metadata": self.metadata,
173 "max_runs_per_component": self._max_runs_per_component,
174 "components": components,
175 "connections": connections,
176 "connection_type_validation": self._connection_type_validation,
177 }
179 @classmethod
180 @mark_deserialization_internal
181 def from_dict(
182 cls: type[T],
183 data: dict[str, Any],
184 callbacks: DeserializationCallbacks | None = None,
185 *,
186 allowed_modules: list[str] | None = None,
187 unsafe: bool = False,
188 **kwargs: Any,
189 ) -> T:
190 """
191 Deserializes the pipeline from a dictionary.
193 :param data:
194 Dictionary to deserialize from.
195 :param callbacks:
196 Callbacks to invoke during deserialization.
197 :param allowed_modules:
198 Additional module patterns whose classes may be imported during deserialization.
199 By default, only modules under `haystack`, `haystack_integrations`, `haystack_experimental`,
200 `builtins`, `typing`, and `collections` are trusted. See
201 `haystack.core.serialization.allow_deserialization_module` for the matching semantics.
202 :param unsafe:
203 If `True`, bypass the deserialization allowlist entirely. Only use this when you fully
204 trust the source of the serialized data — any class in any importable module can be
205 instantiated.
206 :param kwargs:
207 `components`: a dictionary of `{name: instance}` to reuse instances of components instead of creating new
208 ones.
209 :returns:
210 Deserialized component.
211 """
212 with _deserialization_context(allowed_modules=allowed_modules, unsafe=unsafe):
213 return cls._from_dict_impl(data, callbacks, **kwargs)
215 @classmethod
216 def _from_dict_impl(
217 cls: type[T], data: dict[str, Any], callbacks: DeserializationCallbacks | None = None, **kwargs: Any
218 ) -> T:
219 data_copy = _deepcopy_with_exceptions(data) # to prevent modification of original data
220 metadata = data_copy.get("metadata", {})
221 max_runs_per_component = data_copy.get("max_runs_per_component", 100)
222 connection_type_validation = data_copy.get("connection_type_validation", True)
223 pipe = cls(
224 metadata=metadata,
225 max_runs_per_component=max_runs_per_component,
226 connection_type_validation=connection_type_validation,
227 )
228 components_to_reuse = kwargs.get("components", {})
229 for name, component_data in data_copy.get("components", {}).items():
230 if name in components_to_reuse:
231 # Reuse an instance
232 instance = components_to_reuse[name]
233 else:
234 if "type" not in component_data:
235 raise PipelineError(f"Missing 'type' in component '{name}'")
237 component_type = component_data["type"]
238 if isinstance(component_type, str) and "." in component_type:
239 _check_module_allowed(component_type.rsplit(".", 1)[0])
241 if component_type not in component.registry:
242 try:
243 # Import the module first...
244 module, _ = component_type.rsplit(".", 1)
245 logger.debug("Trying to import module {module_name}", module_name=module)
246 type_serialization.thread_safe_import(module)
247 # ...then try again
248 if component_type not in component.registry:
249 raise PipelineError( # noqa: TRY301
250 f"Successfully imported module '{module}' but couldn't find "
251 f"'{component_type}' in the component registry.\n"
252 f"The component might be registered under a different path. "
253 f"Here are the registered components:\n {list(component.registry.keys())}\n"
254 )
255 except (ImportError, PipelineError, ValueError) as e:
256 raise PipelineError(
257 f"Component '{component_type}' (name: '{name}') not imported. Please "
258 f"check that the package is installed and the component path is correct."
259 ) from e
261 # Create a new one
262 component_class = component.registry[component_type]
264 try:
265 instance = component_from_dict(component_class, component_data, name, callbacks)
266 except Exception as e:
267 # Convert to JSON with indentation, truncate if too long
268 try:
269 data_str = json.dumps(component_data, default=str, indent=2)
270 except Exception:
271 data_str = str(component_data)
273 max_len = 1000
274 if len(data_str) > max_len:
275 data_str = data_str[:max_len] + "\n... (truncated)"
277 msg = (
278 f"Couldn't deserialize component '{name}' of class '{component_class.__name__}' "
279 f"with the following data:\n{data_str}\n\n"
280 f"Original error: {e}"
281 )
282 raise DeserializationError(msg) from e
283 pipe.add_component(name=name, instance=instance)
285 for connection in data.get("connections", []):
286 if "sender" not in connection:
287 raise PipelineError(f"Missing sender in connection: {connection}")
288 if "receiver" not in connection:
289 raise PipelineError(f"Missing receiver in connection: {connection}")
290 pipe.connect(sender=connection["sender"], receiver=connection["receiver"])
292 return pipe
294 def dumps(self, marshaller: Marshaller = DEFAULT_MARSHALLER) -> str:
295 """
296 Returns the string representation of this pipeline according to the format dictated by the `Marshaller` in use.
298 :param marshaller:
299 The Marshaller used to create the string representation. Defaults to `YamlMarshaller`.
300 :returns:
301 A string representing the pipeline.
302 """
303 return marshaller.marshal(self.to_dict())
305 def dump(self, fp: TextIO, marshaller: Marshaller = DEFAULT_MARSHALLER) -> None:
306 """
307 Writes the string representation of this pipeline to the file-like object passed in the `fp` argument.
309 :param fp:
310 A file-like object ready to be written to.
311 :param marshaller:
312 The Marshaller used to create the string representation. Defaults to `YamlMarshaller`.
313 """
314 fp.write(marshaller.marshal(self.to_dict()))
316 @classmethod
317 @mark_deserialization_internal
318 def loads(
319 cls: type[T],
320 data: str | bytes | bytearray,
321 marshaller: Marshaller = DEFAULT_MARSHALLER,
322 callbacks: DeserializationCallbacks | None = None,
323 *,
324 allowed_modules: list[str] | None = None,
325 unsafe: bool = False,
326 ) -> T:
327 """
328 Creates a `Pipeline` object from the string representation passed in the `data` argument.
330 :param data:
331 The string representation of the pipeline, can be `str`, `bytes` or `bytearray`.
332 :param marshaller:
333 The Marshaller used to create the string representation. Defaults to `YamlMarshaller`.
334 :param callbacks:
335 Callbacks to invoke during deserialization.
336 :param allowed_modules:
337 Additional module patterns whose classes may be imported during deserialization.
338 By default, only modules under `haystack`, `haystack_integrations`, `haystack_experimental`,
339 `builtins`, `typing`, and `collections` are trusted.
340 :param unsafe:
341 If `True`, bypass the deserialization allowlist entirely. Only use this when you fully
342 trust the source of the serialized data — any class in any importable module can be
343 instantiated.
344 :raises DeserializationError:
345 If an error occurs during deserialization.
346 :returns:
347 A `Pipeline` object.
348 """
349 try:
350 deserialized_data = marshaller.unmarshal(data)
351 except Exception as e:
352 raise DeserializationError(
353 "Error while unmarshalling serialized pipeline data. This is usually "
354 "caused by malformed or invalid syntax in the serialized representation."
355 ) from e
357 return cls.from_dict(deserialized_data, callbacks, allowed_modules=allowed_modules, unsafe=unsafe)
359 @classmethod
360 @mark_deserialization_internal
361 def load(
362 cls: type[T],
363 fp: TextIO,
364 marshaller: Marshaller = DEFAULT_MARSHALLER,
365 callbacks: DeserializationCallbacks | None = None,
366 *,
367 allowed_modules: list[str] | None = None,
368 unsafe: bool = False,
369 ) -> T:
370 """
371 Creates a `Pipeline` object from a string representation.
373 The string representation is read from the file-like object passed in the `fp` argument.
376 :param fp:
377 A file-like object ready to be read from.
378 :param marshaller:
379 The Marshaller used to create the string representation. Defaults to `YamlMarshaller`.
380 :param callbacks:
381 Callbacks to invoke during deserialization.
382 :param allowed_modules:
383 Additional module patterns whose classes may be imported during deserialization.
384 By default, only modules under `haystack`, `haystack_integrations`, `haystack_experimental`,
385 `builtins`, `typing`, and `collections` are trusted.
386 :param unsafe:
387 If `True`, bypass the deserialization allowlist entirely. Only use this when you fully
388 trust the source of the serialized data — any class in any importable module can be
389 instantiated.
390 :raises DeserializationError:
391 If an error occurs during deserialization.
392 :returns:
393 A `Pipeline` object.
394 """
395 return cls.loads(fp.read(), marshaller, callbacks, allowed_modules=allowed_modules, unsafe=unsafe)
397 # Self preserves the concrete subclass for fluent calls, so Pipeline().add_component(...).run() type-checks.
398 def add_component(self, name: str, instance: Component) -> Self:
399 """
400 Add the given component to the pipeline.
402 Components are not connected to anything by default: use `Pipeline.connect()` to connect components together.
403 Component names must be unique, and a component instance can only be added to one pipeline at a time.
404 Adding the same component instance with the same name more than once is a no-op.
406 :param name:
407 The name of the component to add.
408 :param instance:
409 The component instance to add.
410 :returns:
411 The Pipeline instance.
413 :raises ValueError:
414 If a different component with the same name already exists.
415 :raises PipelineValidationError:
416 If the given instance is not a component.
417 :raises PipelineError:
418 If the component instance is already in this pipeline under another name or is in another pipeline.
419 """
420 if not self._validate_component(name, instance):
421 return self
423 self._add_component_to_graph(name, instance)
424 return self
426 def add_components(self, components: dict[str, Component]) -> Self:
427 """
428 Add multiple components to the pipeline.
430 Components are not connected to anything by default: use `Pipeline.connect()` to connect components together.
431 Before adding anything, Haystack checks that every name is valid, every value is a Component instance,
432 no name belongs to a different component, and no instance is assigned to another name or pipeline.
433 If any check fails, the pipeline remains unchanged.
434 Components already present under the same name are ignored when they are the exact same instances.
436 :param components:
437 A dictionary that maps component names to component instances.
438 :returns:
439 The Pipeline instance.
441 :raises ValueError:
442 If a component name is invalid or already belongs to a different component in this pipeline.
443 :raises PipelineValidationError:
444 If one of the given instances is not a component.
445 :raises PipelineError:
446 If a component instance is already in this pipeline under another name, is in another pipeline,
447 or occurs more than once in the mapping.
448 """
449 components_to_add: list[tuple[str, Component]] = []
450 component_names_by_id: dict[int, str] = {}
452 for name, instance in components.items():
453 if not self._validate_component(name, instance):
454 continue
456 instance_id = id(instance)
457 previous_name = component_names_by_id.get(instance_id)
458 if previous_name is not None:
459 raise PipelineError(
460 f"Component instance cannot be added to the pipeline more than once. "
461 f"It is mapped to both '{previous_name}' and '{name}'."
462 )
464 component_names_by_id[instance_id] = name
465 components_to_add.append((name, instance))
467 for name, instance in components_to_add:
468 self._add_component_to_graph(name, instance)
470 return self
472 def _validate_component(self, name: str, instance: Component) -> bool:
473 """Validate a component before adding it, returning whether it needs to be added."""
474 # Component names are unique
475 if name in self.graph.nodes:
476 if self.graph.nodes[name]["instance"] is instance:
477 return False
478 raise ValueError(f"A component named '{name}' already exists in this pipeline: choose another name.")
480 # Components can't be named `_debug`
481 if name == "_debug":
482 raise ValueError("'_debug' is a reserved name for debug output. Choose another name.")
484 # Component names can't have "."
485 if "." in name:
486 raise ValueError(f"{name} is an invalid component name, cannot contain '.' (dot) characters.")
488 # Component instances must be components
489 if not isinstance(instance, Component):
490 raise PipelineValidationError(
491 f"'{type(instance)}' doesn't seem to be a component. Is this class decorated with @component?"
492 )
494 if owning_pipeline := getattr(instance, "__haystack_added_to_pipeline__", None):
495 if owning_pipeline is self:
496 existing_name = self.get_component_name(instance)
497 msg = (
498 f"Component has already been added to this Pipeline under the name '{existing_name}'. "
499 "A component instance can only be added once."
500 )
501 else:
502 msg = (
503 "Component has already been added in another Pipeline. "
504 "Components can't be shared between Pipelines. Create a new instance instead."
505 )
506 raise PipelineError(msg)
508 return True
510 def _add_component_to_graph(self, name: str, instance: Component) -> None:
511 """Add an already validated component to the graph."""
512 setattr(instance, "__haystack_added_to_pipeline__", self) # noqa: B010
513 setattr(instance, "__component_name__", name) # noqa: B010
515 # Add component to the graph, disconnected
516 logger.debug("Adding component '{component_name}' ({component})", component_name=name, component=instance)
517 # We're completely sure the fields exist so we ignore the type error
518 self.graph.add_node(
519 name,
520 instance=instance,
521 input_sockets=instance.__haystack_input__._sockets_dict, # type: ignore[attr-defined]
522 output_sockets=instance.__haystack_output__._sockets_dict, # type: ignore[attr-defined]
523 visits=0,
524 )
526 def remove_component(self, name: str) -> Component:
527 """
528 Remove and returns component from the pipeline.
530 Remove an existing component from the pipeline by providing its name.
531 All edges that connect to the component will also be deleted.
533 :param name:
534 The name of the component to remove.
535 :returns:
536 The removed Component instance.
538 :raises ValueError:
539 If there is no component with that name already in the Pipeline.
540 """
542 # Check that a component with that name is in the Pipeline
543 try:
544 instance = self.get_component(name)
545 except ValueError as exc:
546 raise ValueError(
547 f"There is no component named '{name}' in the pipeline. The valid component names are: ",
548 ", ".join(n for n in self.graph.nodes),
549 ) from exc
551 # Remove this component's name from its neighbors' sockets before the edges are gone,
552 # otherwise the surviving components are left holding dangling references to it.
553 for _, _, edge_data in self.graph.in_edges(name, data=True):
554 sender_socket = edge_data["from_socket"]
555 sender_socket.receivers = [r for r in sender_socket.receivers if r != name]
556 for _, _, edge_data in self.graph.out_edges(name, data=True):
557 receiver_socket = edge_data["to_socket"]
558 receiver_socket.senders = [s for s in receiver_socket.senders if s != name]
559 if (
560 len(receiver_socket.senders) <= 1
561 and receiver_socket.is_lazy_variadic
562 and not receiver_socket.wrap_input_in_list
563 ):
564 receiver_socket.is_lazy_variadic = False
565 receiver_socket.wrap_input_in_list = True
567 # Delete component from the graph, deleting all its connections
568 self.graph.remove_node(name)
570 # Reset the Component sockets' senders and receivers
571 input_sockets = instance.__haystack_input__._sockets_dict # type: ignore[attr-defined]
572 for socket in input_sockets.values():
573 socket.senders = []
574 if socket.is_lazy_variadic and not socket.wrap_input_in_list:
575 socket.is_lazy_variadic = False
576 socket.wrap_input_in_list = True
578 output_sockets = instance.__haystack_output__._sockets_dict # type: ignore[attr-defined]
579 for socket in output_sockets.values():
580 socket.receivers = []
582 # Reset the Component's pipeline reference
583 setattr(instance, "__haystack_added_to_pipeline__", None) # noqa: B010
585 return instance
587 def connect(self, sender: str, receiver: str) -> Self: # noqa: PLR0915 PLR0912 C901
588 """
589 Connects two components together.
591 All components to connect must exist in the pipeline.
592 If connecting to a component that has several output connections, specify the inputs and output names as
593 'component_name.connections_name'.
595 If multiple senders are connected to the same list-typed receiver socket, the socket is
596 promoted to a lazy variadic socket so it can accept all incoming values. With the synchronous
597 `run`, the resulting list is ordered alphabetically by sender component name, not by the order in
598 which `connect()` was called. With the asynchronous run path (`run_async`), no ordering is
599 guaranteed, since components in different branches may run in parallel.
601 :param sender:
602 The component that delivers the value. This can be either just a component name or can be
603 in the format `component_name.connection_name` if the component has multiple outputs.
604 :param receiver:
605 The component that receives the value. This can be either just a component name or can be
606 in the format `component_name.connection_name` if the component has multiple inputs.
608 :returns:
609 The Pipeline instance.
611 :raises PipelineConnectError:
612 If the two components cannot be connected (for example if one of the components is
613 not present in the pipeline, or the connections don't match by type, and so on).
614 """
615 # Edges may be named explicitly by passing 'node_name.edge_name' to connect().
616 sender_component_name, sender_socket_name = parse_connect_string(sender)
617 receiver_component_name, receiver_socket_name = parse_connect_string(receiver)
619 if sender_component_name == receiver_component_name:
620 raise PipelineConnectError("Connecting a Component to itself is not supported.")
622 # Get the nodes data.
623 try:
624 sender_sockets = self.graph.nodes[sender_component_name]["output_sockets"]
625 except KeyError as exc:
626 raise ValueError(f"Component named {sender_component_name} not found in the pipeline.") from exc
627 try:
628 receiver_sockets = self.graph.nodes[receiver_component_name]["input_sockets"]
629 except KeyError as exc:
630 raise ValueError(f"Component named {receiver_component_name} not found in the pipeline.") from exc
632 if not sender_sockets:
633 raise PipelineConnectError(
634 f"'{sender_component_name}' does not have any output connections. "
635 f"Please check that the output types of '{sender_component_name}.run' are set, "
636 f"for example by using the '@component.output_types' decorator."
637 )
639 # If the name of either socket is given, get the socket
640 sender_socket: OutputSocket | None = None
641 if sender_socket_name:
642 sender_socket = sender_sockets.get(sender_socket_name)
643 if not sender_socket:
644 raise PipelineConnectError(
645 f"'{sender}' does not exist. "
646 f"Output connections of {sender_component_name} are: "
647 + ", ".join([f"{name} (type {_type_name(socket.type)})" for name, socket in sender_sockets.items()])
648 )
650 receiver_socket: InputSocket | None = None
651 if receiver_socket_name:
652 receiver_socket = receiver_sockets.get(receiver_socket_name)
653 if not receiver_socket:
654 raise PipelineConnectError(
655 f"'{receiver} does not exist. "
656 f"Input connections of {receiver_component_name} are: "
657 + ", ".join(
658 [f"{name} (type {_type_name(socket.type)})" for name, socket in receiver_sockets.items()]
659 )
660 )
662 # Look for a matching connection among the possible ones.
663 # Note that if there is more than one possible connection but two sockets match by name, they're paired.
664 sender_socket_candidates: list[OutputSocket] = (
665 [sender_socket] if sender_socket else list(sender_sockets.values())
666 )
667 receiver_socket_candidates: list[InputSocket] = (
668 [receiver_socket] if receiver_socket else list(receiver_sockets.values())
669 )
671 conversion_strategy = None
673 # Find all possible connections between these two components
674 possible_connections: list[tuple[OutputSocket, InputSocket, ConversionStrategyType]] = []
675 for sender_sock, receiver_sock in itertools.product(sender_socket_candidates, receiver_socket_candidates):
676 is_compat, conversion_strategy = _types_are_compatible(
677 sender_sock.type, receiver_sock.type, self._connection_type_validation
678 )
679 if is_compat:
680 possible_connections.append((sender_sock, receiver_sock, conversion_strategy))
682 # If there are multiple possibilities, prioritize strict matches over convertible ones.
683 # This ensures backward compatibility: previously, pipelines did not allow type conversion.
684 if len(possible_connections) > 1 and self._connection_type_validation:
685 strict_matches = [
686 (out_sock, in_sock, None)
687 for out_sock, in_sock, conversion_strategy in possible_connections
688 if conversion_strategy is None
689 ]
690 if strict_matches:
691 possible_connections[:] = strict_matches
693 # We need this status for error messages, since we might need it in multiple places we calculate it here
694 status = _connections_status(
695 sender_node=sender_component_name,
696 sender_sockets=sender_socket_candidates,
697 receiver_node=receiver_component_name,
698 receiver_sockets=receiver_socket_candidates,
699 )
701 if not possible_connections:
702 # There's no possible connection between these two components
703 if len(sender_socket_candidates) == len(receiver_socket_candidates) == 1:
704 msg = (
705 f"Cannot connect '{sender_component_name}.{sender_socket_candidates[0].name}' with "
706 f"'{receiver_component_name}.{receiver_socket_candidates[0].name}': "
707 f"their declared input and output types do not match.\n{status}"
708 )
709 else:
710 msg = (
711 f"Cannot connect '{sender_component_name}' with '{receiver_component_name}': "
712 f"no matching connections available.\n{status}"
713 )
714 raise PipelineConnectError(msg)
716 if len(possible_connections) == 1:
717 # There's only one possible connection, use it
718 sender_socket = possible_connections[0][0]
719 receiver_socket = possible_connections[0][1]
720 conversion_strategy = possible_connections[0][2]
722 if len(possible_connections) > 1:
723 # There are multiple possible connection, let's try to match them by name
724 name_matches = [
725 (out_sock, in_sock, conversion_strategy_)
726 for out_sock, in_sock, conversion_strategy_ in possible_connections
727 if in_sock.name == out_sock.name
728 ]
729 if len(name_matches) != 1:
730 # There's are either no matches or more than one, we can't pick one reliably
731 msg = (
732 f"Cannot connect '{sender_component_name}' with "
733 f"'{receiver_component_name}': more than one connection is possible "
734 "between these components. Please specify the connection name, like: "
735 f"pipeline.connect('{sender_component_name}.{possible_connections[0][0].name}', "
736 f"'{receiver_component_name}.{possible_connections[0][1].name}').\n{status}"
737 )
738 raise PipelineConnectError(msg)
740 # Get the only possible match
741 sender_socket = name_matches[0][0]
742 receiver_socket = name_matches[0][1]
743 conversion_strategy = name_matches[0][2]
745 # Connection must be valid on both sender/receiver sides
746 if not sender_socket or not receiver_socket or not sender_component_name or not receiver_component_name:
747 if sender_component_name and sender_socket:
748 sender_repr = f"{sender_component_name}.{sender_socket.name} ({_type_name(sender_socket.type)})"
749 else:
750 sender_repr = "input needed"
752 if receiver_component_name and receiver_socket:
753 receiver_repr = f"({_type_name(receiver_socket.type)}) {receiver_component_name}.{receiver_socket.name}"
754 else:
755 receiver_repr = "output"
756 msg = f"Connection must have both sender and receiver: {sender_repr} -> {receiver_repr}"
757 raise PipelineConnectError(msg)
759 logger.debug(
760 "Connecting '{sender_component}.{sender_socket_name}' to '{receiver_component}.{receiver_socket_name}'",
761 sender_component=sender_component_name,
762 sender_socket_name=sender_socket.name,
763 receiver_component=receiver_component_name,
764 receiver_socket_name=receiver_socket.name,
765 )
767 if receiver_component_name in sender_socket.receivers and sender_component_name in receiver_socket.senders:
768 # This is already connected, nothing to do
769 return self
771 if receiver_socket.senders:
772 receiver_socket = self._make_socket_auto_variadic(
773 component_name=receiver_component_name, receiver_socket=receiver_socket, error_type=PipelineConnectError
774 )
776 # Update the sockets with the new connection
777 sender_socket.receivers.append(receiver_component_name)
778 receiver_socket.senders.append(sender_component_name)
780 # Create the new connection
781 self.graph.add_edge(
782 sender_component_name,
783 receiver_component_name,
784 key=f"{sender_socket.name}/{receiver_socket.name}",
785 conn_type=_type_name(sender_socket.type),
786 from_socket=sender_socket,
787 to_socket=receiver_socket,
788 mandatory=receiver_socket.is_mandatory,
789 conversion_strategy=conversion_strategy,
790 )
791 return self
793 def connect_many(self, connections: list[tuple[str, str]]) -> Self:
794 """
795 Connect multiple pairs of components.
797 Connections are made in the order provided. If connecting a pair raises an exception, no subsequent pairs
798 are connected, while earlier successful connections remain in the pipeline. Repeating an existing connection
799 is a no-op.
801 :param connections:
802 A list of `(sender, receiver)` pairs. Each value uses the same format accepted by
803 `Pipeline.connect()`.
804 :returns:
805 The Pipeline instance.
807 :raises PipelineConnectError:
808 If a pair of components cannot be connected.
809 :raises ValueError:
810 If a sender or receiver component is not present in the pipeline.
811 """
812 for sender, receiver in connections:
813 self.connect(sender, receiver)
815 return self
817 def get_component(self, name: str) -> Component:
818 """
819 Get the component with the specified name from the pipeline.
821 :param name:
822 The name of the component.
823 :returns:
824 The instance of that component.
826 :raises ValueError:
827 If a component with that name is not present in the pipeline.
828 """
829 try:
830 return self.graph.nodes[name]["instance"]
831 except KeyError as exc:
832 raise ValueError(f"Component named {name} not found in the pipeline.") from exc
834 def get_component_name(self, instance: Component) -> str:
835 """
836 Returns the name of the Component instance if it has been added to this Pipeline or an empty string otherwise.
838 :param instance:
839 The Component instance to look for.
840 :returns:
841 The name of the Component instance.
842 """
843 for name, inst in self.graph.nodes(data="instance"):
844 if inst == instance:
845 return name
846 return ""
848 def inputs(self, include_components_with_connected_inputs: bool = False) -> dict[str, dict[str, Any]]:
849 """
850 Returns a dictionary containing the inputs of a pipeline.
852 Each key in the dictionary corresponds to a component name, and its value is another dictionary that describes
853 the input sockets of that component, including their types and whether they are optional.
855 :param include_components_with_connected_inputs:
856 If `False`, only components that have disconnected input edges are
857 included in the output.
858 :returns:
859 A dictionary where each key is a pipeline component name and each value is a dictionary of
860 inputs sockets of that component.
861 """
862 inputs: dict[str, dict[str, Any]] = {}
863 for component_name, data in find_pipeline_inputs(self.graph, include_components_with_connected_inputs).items():
864 sockets_description = {}
865 for socket in data:
866 # Variadic mandatory sockets with existing connections don't require user input, so treat them as
867 # optional.
868 is_mandatory = socket.is_mandatory and not socket.senders
869 sockets_description[socket.name] = {"type": socket.type, "is_mandatory": is_mandatory}
870 if not socket.is_mandatory:
871 sockets_description[socket.name]["default_value"] = socket.default_value
873 if sockets_description:
874 inputs[component_name] = sockets_description
875 return inputs
877 def outputs(self, include_components_with_connected_outputs: bool = False) -> dict[str, dict[str, Any]]:
878 """
879 Returns a dictionary containing the outputs of a pipeline.
881 Each key in the dictionary corresponds to a component name, and its value is another dictionary that describes
882 the output sockets of that component.
884 :param include_components_with_connected_outputs:
885 If `False`, only components that have disconnected output edges are
886 included in the output.
887 :returns:
888 A dictionary where each key is a pipeline component name and each value is a dictionary of
889 output sockets of that component.
890 """
891 return {
892 comp: {socket.name: {"type": socket.type} for socket in data}
893 for comp, data in find_pipeline_outputs(self.graph, include_components_with_connected_outputs).items()
894 if data
895 }
897 def show(
898 self,
899 *,
900 server_url: str = "https://mermaid.ink",
901 params: dict | None = None,
902 timeout: int = 30,
903 super_component_expansion: bool = False,
904 ) -> None:
905 """
906 Display an image representing this `Pipeline` in a Jupyter notebook.
908 This function generates a diagram of the `Pipeline` using a Mermaid server and displays it directly in
909 the notebook.
911 :param server_url:
912 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
913 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
914 info on how to set up your own Mermaid server.
916 :param params:
917 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
918 Supported keys:
919 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
920 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
921 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
922 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
923 - width: Width of the output image (integer).
924 - height: Height of the output image (integer).
925 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
926 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
927 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
928 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
930 :param timeout:
931 Timeout in seconds for the request to the Mermaid server.
933 :param super_component_expansion:
934 If set to True and the pipeline contains SuperComponents the diagram will show the internal structure of
935 super-components as if they were components part of the pipeline instead of a "black-box".
936 Otherwise, only the super-component itself will be displayed.
938 :raises PipelineDrawingError:
939 If the function is called outside of a Jupyter notebook or if there is an issue with rendering.
940 """
942 if is_in_jupyter():
943 from IPython.display import Image, display
945 if super_component_expansion:
946 graph, super_component_mapping = self._merge_super_component_pipelines()
947 else:
948 graph = self.graph
949 super_component_mapping = None
951 image_data = _to_mermaid_image(
952 graph,
953 server_url=server_url,
954 params=params,
955 timeout=timeout,
956 super_component_mapping=super_component_mapping,
957 )
958 display(Image(image_data))
959 else:
960 msg = "This method is only supported in Jupyter notebooks. Use Pipeline.draw() to save an image locally."
961 raise PipelineDrawingError(msg)
963 def draw(
964 self,
965 *,
966 path: Path,
967 server_url: str = "https://mermaid.ink",
968 params: dict | None = None,
969 timeout: int = 30,
970 super_component_expansion: bool = False,
971 ) -> None:
972 """
973 Save an image representing this `Pipeline` to the specified file path.
975 This function generates a diagram of the `Pipeline` using the Mermaid server and saves it to the provided path.
977 :param path:
978 The file path where the generated image will be saved.
980 :param server_url:
981 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
982 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
983 info on how to set up your own Mermaid server.
985 :param params:
986 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
987 Supported keys:
988 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
989 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
990 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
991 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
992 - width: Width of the output image (integer).
993 - height: Height of the output image (integer).
994 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
995 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
996 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
997 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
999 :param timeout:
1000 Timeout in seconds for the request to the Mermaid server.
1002 :param super_component_expansion:
1003 If set to True and the pipeline contains SuperComponents the diagram will show the internal structure of
1004 super-components as if they were components part of the pipeline instead of a "black-box".
1005 Otherwise, only the super-component itself will be displayed.
1007 :raises PipelineDrawingError:
1008 If there is an issue with rendering or saving the image.
1009 """
1011 # Before drawing we edit a bit the graph, to avoid modifying the original that is
1012 # used for running the pipeline we copy it.
1013 if super_component_expansion:
1014 graph, super_component_mapping = self._merge_super_component_pipelines()
1015 else:
1016 graph = self.graph
1017 super_component_mapping = None
1019 image_data = _to_mermaid_image(
1020 graph,
1021 server_url=server_url,
1022 params=params,
1023 timeout=timeout,
1024 super_component_mapping=super_component_mapping,
1025 )
1026 Path(path).write_bytes(image_data)
1028 def walk(self) -> Iterator[tuple[str, Component]]:
1029 """
1030 Visits each component in the pipeline exactly once and yields its name and instance.
1032 No guarantees are provided on the visiting order.
1034 :returns:
1035 An iterator of tuples of component name and component instance.
1036 """
1037 for component_name, instance in self.graph.nodes(data="instance"): # noqa: UP028
1038 yield component_name, instance
1040 def warm_up(self) -> None:
1041 """
1042 Make sure all components are warm.
1044 It's the component's responsibility to make sure this method can be called at every `Pipeline.run()`
1045 without re-initializing everything.
1046 """
1047 for component_name in self.graph.nodes:
1048 if hasattr(self.graph.nodes[component_name]["instance"], "warm_up"):
1049 logger.info("Warming up component {component_name}...", component_name=component_name)
1050 self.graph.nodes[component_name]["instance"].warm_up()
1052 async def warm_up_async(self) -> None:
1053 """
1054 Make sure all components are warm, using the async warm-up path where available.
1056 Each component is warmed up with `warm_up_async` if it has one, otherwise with its sync `warm_up`.
1057 Both run on the event loop, never offloaded to a worker thread.
1058 This ensures that if an async client is created during `warm-up` (residual scenario), it binds to the loop that
1059 `run_async` will use.
1060 """
1061 for component_name in self.graph.nodes:
1062 instance = self.graph.nodes[component_name]["instance"]
1063 if hasattr(instance, "warm_up_async"):
1064 logger.info("Warming up component {component_name}...", component_name=component_name)
1065 await instance.warm_up_async()
1066 elif hasattr(instance, "warm_up"):
1067 logger.info("Warming up component {component_name}...", component_name=component_name)
1068 instance.warm_up()
1070 def close(self) -> None:
1071 """
1072 Release resources held by the pipeline's components by calling each component's `close` method.
1074 Only the synchronous side of each component is released here; use `close_async` to release async clients.
1075 """
1076 for component_name in self.graph.nodes:
1077 instance = self.graph.nodes[component_name]["instance"]
1078 if hasattr(instance, "close"):
1079 logger.info("Closing component {component_name}...", component_name=component_name)
1080 instance.close()
1082 async def close_async(self) -> None:
1083 """
1084 Release resources held by the pipeline's components, using the async close path where available.
1086 Each component is closed with `close_async` if it has one, otherwise with its sync `close`.
1087 """
1088 for component_name in self.graph.nodes:
1089 instance = self.graph.nodes[component_name]["instance"]
1090 if hasattr(instance, "close_async"):
1091 logger.info("Closing component {component_name}...", component_name=component_name)
1092 await instance.close_async()
1093 elif hasattr(instance, "close"):
1094 logger.info("Closing component {component_name}...", component_name=component_name)
1095 instance.close()
1097 @staticmethod
1098 def _create_component_span(
1099 component_name: str, instance: Component, inputs: dict[str, Any], parent_span: tracing.Span | None = None
1100 ) -> ContextManager[tracing.Span]:
1101 return tracing.tracer.trace(
1102 "haystack.component.run",
1103 tags={
1104 "haystack.component.name": component_name,
1105 "haystack.component.type": instance.__class__.__name__,
1106 "haystack.component.fully_qualified_type": generate_qualified_class_name(type(instance)),
1107 "haystack.component.input_types": {k: type(v).__name__ for k, v in inputs.items()},
1108 "haystack.component.input_spec": {
1109 key: {
1110 "type": value.type.__name__ if type(value.type) is type else str(value.type),
1111 "senders": value.senders,
1112 }
1113 for key, value in instance.__haystack_input__._sockets_dict.items() # type: ignore
1114 },
1115 "haystack.component.output_spec": {
1116 key: {
1117 "type": value.type.__name__ if type(value.type) is type else str(value.type),
1118 "receivers": value.receivers,
1119 }
1120 for key, value in instance.__haystack_output__._sockets_dict.items() # type: ignore
1121 },
1122 },
1123 parent_span=parent_span,
1124 )
1126 def validate_input(self, data: dict[str, Any]) -> None:
1127 """
1128 Validates pipeline input data.
1130 Validates that data:
1131 * Each Component name actually exists in the Pipeline
1132 * Each Component is not missing any input
1133 * Each Component has only one input per input socket, if not variadic
1134 * Each Component doesn't receive inputs that are already sent by another Component
1136 :param data:
1137 A dictionary of inputs for the pipeline's components. Each key is a component name.
1139 :raises ValueError:
1140 If inputs are invalid according to the above.
1141 """
1142 for component_name, component_inputs in data.items():
1143 # Check that the component exists
1144 if component_name not in self.graph.nodes:
1145 raise ValueError(f"Component named '{component_name}' not found in the pipeline.")
1146 # Check that no input is provided that the component can't accept
1147 instance = self.graph.nodes[component_name]["instance"]
1148 for input_name in component_inputs.keys():
1149 if input_name not in instance.__haystack_input__._sockets_dict:
1150 raise ValueError(f"Input '{input_name}' not found in component '{component_name}'.")
1152 for component_name in self.graph.nodes:
1153 instance = self.graph.nodes[component_name]["instance"]
1154 for socket_name, socket in instance.__haystack_input__._sockets_dict.items():
1155 component_inputs = data.get(component_name, {})
1156 # Check that no mandatory input is missing for any component in the pipeline
1157 if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
1158 raise ValueError(f"Missing mandatory input '{socket_name}' for component '{component_name}'.")
1160 # Check if an input is provided more than once for non-variadic sockets
1161 if socket.senders and socket_name in component_inputs:
1162 self._make_socket_auto_variadic(
1163 component_name=component_name, receiver_socket=socket, error_type=ValueError
1164 )
1166 def _make_socket_auto_variadic(
1167 self, component_name: str, receiver_socket: InputSocket, error_type: type[Exception]
1168 ) -> InputSocket:
1169 """
1170 Attempts to make the receiver socket lazy variadic in-place to accommodate a new sender.
1172 A socket is automatically made lazy variadic when:
1173 - It already has at least one connected sender
1174 - It is not already variadic
1175 - Its type is list, Optional[list], a union of list types
1177 When auto-variadicity is applied, `wrap_input_in_list` is also set to False so that sender output types match
1178 the receiver socket's declared list type directly.
1180 :param component_name:
1181 Name of the component owning the receiver socket, used in error messages.
1182 :param receiver_socket:
1183 The receiver socket to inspect and potentially modify in-place.
1184 :param error_type:
1185 Exception class to raise on failure (e.g. ValueError or PipelineConnectError).
1186 :returns:
1187 The (possibly modified) receiver socket.
1188 :raises error_type:
1189 If the socket cannot accept multiple senders given its type constraints.
1190 """
1191 # If it's already variadic, we return as-is
1192 if receiver_socket.is_variadic:
1193 return receiver_socket
1195 # Get receiver origin
1196 receiver_origin = _safe_get_origin(receiver_socket.type)
1198 # Handle Union types
1199 if receiver_origin == Union:
1200 # Unwrap Optional types
1201 non_none_args = [a for a in get_args(receiver_socket.type) if a is not type(None)]
1202 if len(non_none_args) == 1:
1203 receiver_origin = _safe_get_origin(non_none_args[0])
1204 # Handle Union of list types (e.g. list[int] | list[str])
1205 elif all(_safe_get_origin(arg) == list for arg in non_none_args):
1206 receiver_origin = list
1208 # If the receiver origin is a list, we can make the socket lazy variadic
1209 if receiver_origin == list:
1210 receiver_socket.is_lazy_variadic = True
1211 receiver_socket.wrap_input_in_list = False
1212 return receiver_socket
1214 raise error_type(
1215 f"Component '{component_name}' cannot accept multiple inputs to '{receiver_socket.name}'. "
1216 f"It is already connected to component '{receiver_socket.senders[0]}', and it can only accept "
1217 f"inputs from multiple senders if its type is list, Optional[list], or union of list types."
1218 )
1220 def _prepare_component_input_data(self, data: dict[str, Any]) -> dict[str, dict[str, Any]]:
1221 """
1222 Prepares input data for pipeline components.
1224 Organizes input data for pipeline components and identifies any inputs that are not matched to any
1225 component's input slots. Deep-copies data items to avoid sharing mutables across multiple components.
1227 This method processes a flat dictionary of input data, where each key-value pair represents an input name
1228 and its corresponding value. It distributes these inputs to the appropriate pipeline components based on
1229 their input requirements. Inputs that don't match any component's input slots are classified as unresolved.
1231 :param data:
1232 A dictionary potentially having input names as keys and input values as values.
1234 :returns:
1235 A dictionary mapping component names to their respective matched inputs.
1236 """
1237 # check whether the data is a nested dictionary of component inputs where each key is a component name
1238 # and each value is a dictionary of input parameters for that component
1239 is_nested_component_input = all(isinstance(value, dict) for value in data.values())
1240 if not is_nested_component_input:
1241 # flat input, a dict where keys are input names and values are the corresponding values
1242 # we need to convert it to a nested dictionary of component inputs and then run the pipeline
1243 # just like in the previous case
1244 pipeline_input_data: dict[str, dict[str, Any]] = defaultdict(dict)
1245 unresolved_kwargs = {}
1247 # Retrieve the input slots for each component in the pipeline
1248 available_inputs: dict[str, dict[str, Any]] = self.inputs()
1250 # Go through all provided to distribute them to the appropriate component inputs
1251 for input_name, input_value in data.items():
1252 resolved_at_least_once = False
1254 # Check each component to see if it has a slot for the current kwarg
1255 for component_name, component_inputs in available_inputs.items():
1256 if input_name in component_inputs:
1257 # If a match is found, add the kwarg to the component's input data
1258 pipeline_input_data[component_name][input_name] = input_value
1259 resolved_at_least_once = True
1261 if not resolved_at_least_once:
1262 unresolved_kwargs[input_name] = input_value
1264 if unresolved_kwargs:
1265 logger.warning(
1266 "Inputs {input_keys} were not matched to any component inputs, please check your run parameters.",
1267 input_keys=list(unresolved_kwargs.keys()),
1268 )
1270 data = dict(pipeline_input_data)
1272 # deepcopying the inputs prevents the Pipeline run logic from being altered unexpectedly
1273 # when the same input reference is passed to multiple components.
1274 for component_name, component_inputs in data.items():
1275 data[component_name] = {k: _deepcopy_with_exceptions(v) for k, v in component_inputs.items()}
1277 return data
1279 def _find_receivers_from(
1280 self, component_name: str
1281 ) -> list[tuple[str, OutputSocket, InputSocket, ConversionStrategyType]]:
1282 """
1283 Utility function to find all Components that receive input from `component_name`.
1285 :param component_name:
1286 Name of the sender Component
1288 :returns: A list of tuples containing:
1289 - receiver component name
1290 - sender OutputSocket
1291 - receiver InputSocket
1292 - ConversionStrategy if conversion is required, otherwise None.
1293 """
1294 res = []
1295 for _, receiver_name, connection in self.graph.edges(nbunch=component_name, data=True):
1296 sender_socket: OutputSocket = connection["from_socket"]
1297 receiver_socket: InputSocket = connection["to_socket"]
1298 res.append((receiver_name, sender_socket, receiver_socket, connection.get("conversion_strategy")))
1299 return res
1301 @staticmethod
1302 def _convert_to_internal_format(pipeline_inputs: dict[str, Any]) -> InputsType:
1303 """
1304 Converts the inputs to the pipeline to the format that is needed for the internal `Pipeline.run` logic.
1306 Example Input:
1307 {'prompt_builder': {'question': 'Who lives in Paris?'}, 'retriever': {'query': 'Who lives in Paris?'}}
1308 Example Output:
1309 {'prompt_builder': {'question': [{'sender': None, 'value': 'Who lives in Paris?'}]},
1310 'retriever': {'query': [{'sender': None, 'value': 'Who lives in Paris?'}]}}
1312 :param pipeline_inputs: Inputs to the pipeline.
1313 :returns: Converted inputs that can be used by the internal `Pipeline.run` logic.
1314 """
1315 inputs: InputsType = {}
1316 for component_name, socket_dict in pipeline_inputs.items():
1317 inputs[component_name] = {}
1318 for socket_name, value in socket_dict.items():
1319 inputs[component_name][socket_name] = [{"sender": None, "value": value}]
1321 return inputs
1323 @staticmethod
1324 def _consume_component_inputs(
1325 component_name: str, component: dict, inputs: InputsType, is_resume: bool = False
1326 ) -> dict[str, Any]:
1327 """
1328 Extracts the inputs needed to run for the component and removes them from the global inputs state.
1330 :param component_name: The name of a component.
1331 :param component: Component with component metadata.
1332 :param inputs: Global inputs state.
1333 :param is_resume: Whether the component is being resumed from a breakpoint. If True, the inputs
1334 have already been consumed, so the first available value is returned for each socket.
1335 :returns: The inputs for the component.
1336 """
1337 component_inputs = inputs.get(component_name, {})
1338 consumed_inputs = {}
1339 greedy_inputs_to_remove = set()
1340 for socket_name, socket in component["input_sockets"].items():
1341 socket_inputs = component_inputs.get(socket_name, [])
1342 socket_inputs_values = [
1343 sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _NoOutputProduced)
1344 ]
1346 # if we are resuming a component, the inputs are already consumed, so we just return the first input
1347 if is_resume:
1348 consumed_inputs[socket_name] = socket_inputs_values[0]
1349 continue
1351 if socket_inputs_values:
1352 if socket.is_greedy:
1353 # We need to keep track of greedy inputs because we always remove them, even if they come from
1354 # outside the pipeline. Otherwise, a greedy input from the user would trigger a pipeline to run
1355 # indefinitely.
1356 greedy_inputs_to_remove.add(socket_name)
1357 consumed_inputs[socket_name] = [socket_inputs_values[0]]
1358 elif socket.is_lazy_variadic:
1359 if socket.wrap_input_in_list:
1360 # We use all inputs provided to the socket on a lazy variadic socket.
1361 # So keep it wrapped in a list.
1362 consumed_inputs[socket_name] = socket_inputs_values
1363 else:
1364 # We flatten one-level of lists for lazy variadic sockets that don't wrap inputs in lists.
1365 # This way the incoming inputs match the expected type of the socket.
1366 consumed_inputs[socket_name] = list(itertools.chain.from_iterable(socket_inputs_values))
1367 else:
1368 # For a normal socket we only care about the first input provided to the socket.
1369 consumed_inputs[socket_name] = socket_inputs_values[0]
1371 # We prune all inputs except for those that were provided from outside the pipeline (e.g. user inputs).
1372 pruned_inputs = {
1373 socket_name: [
1374 sock for sock in socket if sock["sender"] is None and socket_name not in greedy_inputs_to_remove
1375 ]
1376 for socket_name, socket in component_inputs.items()
1377 }
1378 pruned_inputs = {socket_name: socket for socket_name, socket in pruned_inputs.items() if len(socket) > 0}
1380 inputs[component_name] = pruned_inputs
1382 return consumed_inputs
1384 def _fill_queue(
1385 self, component_names: list[str], inputs: InputsType, component_visits: dict[str, int]
1386 ) -> FIFOPriorityQueue:
1387 """
1388 Calculates the execution priority for each component and inserts it into the priority queue.
1390 :param component_names: Names of the components to put into the queue.
1391 :param inputs: Inputs to the components.
1392 :param component_visits: Current state of component visits.
1393 :returns: A prioritized queue of component names.
1394 """
1395 priority_queue = FIFOPriorityQueue()
1396 for component_name in component_names:
1397 comp = self._get_component_with_graph_metadata_and_visits(component_name, component_visits[component_name])
1398 priority = self._calculate_priority(comp, inputs.get(component_name, {}))
1399 priority_queue.push(component_name, priority)
1400 return priority_queue
1402 @staticmethod
1403 def _calculate_priority(comp: dict, comp_inputs: dict[str, list[dict[str, Any]]]) -> ComponentPriority:
1404 """
1405 Calculates the execution priority for a component depending on the component's inputs.
1407 :param comp: Component metadata and component instance.
1408 :param comp_inputs: Inputs to the component.
1409 :returns: Priority value for the component.
1410 """
1411 # NOTE: Even if a component can run, it doesn't mean it's ready to run since it could be waiting for optional
1412 # inputs. This is why it's only used to determine if a component is BLOCKED or not.
1413 if not can_component_run(comp, comp_inputs):
1414 return ComponentPriority.BLOCKED
1415 if is_any_greedy_socket_ready(comp, comp_inputs) and are_all_sockets_ready(comp, comp_inputs):
1416 # This priority is explicitly used in the async run path + implicitly in _is_queue_stale
1417 # Implicit b/c it checks via ">" operator if there is a component with HIGHEST priority
1418 return ComponentPriority.HIGHEST
1419 if all_predecessors_executed(comp, comp_inputs):
1420 # This priority is explicitly used in the async run path + in _is_queue_stale
1421 return ComponentPriority.READY
1422 # If we make it here it means the component can run but is waiting for more inputs, so we give it the lowest
1423 # priority. This way, components that are ready to run will be prioritized over ones assigned with this prio.
1424 return ComponentPriority.DEFER
1426 def _get_component_with_graph_metadata_and_visits(self, component_name: str, visits: int) -> dict[str, Any]:
1427 """
1428 Returns the component instance alongside input/output-socket metadata from the graph and adds current visits.
1430 We can't store visits in the pipeline graph because this would prevent reentrance / thread-safe execution.
1432 :param component_name: The name of the component.
1433 :param visits: Number of visits for the component.
1434 :returns: dict with keys:
1435 - instance: the component instance
1436 - input_sockets: the component input sockets metadata from the graph
1437 - output_sockets: the component output sockets metadata from the graph
1438 """
1439 comp_dict = self.graph.nodes[component_name]
1440 # We inject the visits into the component dict here to avoid storing it in the graph, which would prevent
1441 # thread-safe execution
1442 return {**comp_dict, "visits": visits}
1444 def _get_next_runnable_component(
1445 self, priority_queue: FIFOPriorityQueue, component_visits: dict[str, int]
1446 ) -> tuple[ComponentPriority, str, dict[str, Any]] | None:
1447 """
1448 Returns the next runnable component alongside its metadata from the priority queue.
1450 :param priority_queue: Priority queue of component names.
1451 :param component_visits: Current state of component visits.
1452 :returns: The next runnable component, the component name, and its priority
1453 or None if no component in the queue can run.
1454 :raises: PipelineMaxComponentRuns if the next runnable component has exceeded the maximum number of runs.
1455 """
1456 item = priority_queue.get()
1458 # If no component is runnable, return None
1459 if item is None:
1460 return None
1462 component_name = item[1]
1463 comp = self._get_component_with_graph_metadata_and_visits(component_name, component_visits[component_name])
1464 # Only raise the max run count error if the component is not blocked, since if it's blocked it means it
1465 # can't run anyway.
1466 if item[0] < ComponentPriority.BLOCKED and comp["visits"] >= self._max_runs_per_component:
1467 msg = f"Maximum run count {self._max_runs_per_component} reached for component '{component_name}'"
1468 raise PipelineMaxComponentRuns(msg)
1469 return ComponentPriority(item[0]), component_name, comp
1471 @staticmethod
1472 def _add_missing_input_defaults(
1473 component_inputs: dict[str, list[dict[str, Any]]], component_input_sockets: dict[str, InputSocket]
1474 ) -> dict[str, Any]:
1475 """
1476 Updates the inputs with the default values for the inputs that are missing
1478 :param component_inputs: Inputs for the component.
1479 :param component_input_sockets: Input sockets of the component.
1480 """
1481 for name, socket in component_input_sockets.items():
1482 if not socket.is_mandatory and name not in component_inputs:
1483 # NOTE: Variadic inputs expect a single default value in the function signature that matches the inner
1484 # type, for example Variadic[str] = "default". When executed inside a pipeline, we wrap this
1485 # default into a list, resulting in ["default"], which is the intended behavior.
1486 #
1487 # However, when the component is executed directly by calling run(), the default is not wrapped and
1488 # is treated as an iterable.
1489 # For strings, this would produce ["d", "e", "f", ...] instead of ["default"].
1490 if socket.is_variadic:
1491 component_inputs[name] = [socket.default_value]
1492 else:
1493 component_inputs[name] = socket.default_value
1495 return component_inputs
1497 def _topological_sort(self) -> dict[str, int]:
1498 """
1499 Returns a topological sort of the components in the pipeline.
1501 If the graph is a DAG, we use lexicographical topological sort to get a deterministic order of the components.
1502 If the graph is not a DAG, we use the condensation of the graph to get a topological sort of the strongly
1503 connected components. This way, components that are part of the same cycle will have the same priority and will
1504 be tie-broken by their name in lexicographical order, while components that are not part of the same cycle will
1505 be tie-broken by their topological order.
1507 :returns:
1508 A dictionary mapping component names to their position in the topological sort.
1509 """
1510 if networkx.is_directed_acyclic_graph(self.graph):
1511 topological_sort = networkx.lexicographical_topological_sort(self.graph)
1512 return {node: idx for idx, node in enumerate(topological_sort)}
1513 else:
1514 condensed = networkx.condensation(self.graph)
1515 condensed_sorted = {node: idx for idx, node in enumerate(networkx.topological_sort(condensed))}
1516 return {
1517 component_name: condensed_sorted[node] for component_name, node in condensed.graph["mapping"].items()
1518 }
1520 def _tiebreak_waiting_components(
1521 self,
1522 component_name: str,
1523 priority: ComponentPriority,
1524 priority_queue: FIFOPriorityQueue,
1525 topological_sort: dict[str, int] | None,
1526 ) -> tuple[str, dict[str, int] | None]:
1527 """
1528 Decides which component to run when multiple components are waiting for inputs with the same priority.
1530 NOTE: This was designed to only tie-break for components with the priority DEFER. Since this function also
1531 removes these components from the priority queue we rely on _is_queue_stale to then refill the priority queue.
1532 And _is_queue_stale only triggers when all remaining components have BLOCKED priority.
1534 :param component_name: The name of the component.
1535 :param priority: Priority of the component.
1536 :param priority_queue: Priority queue of component names.
1537 :param topological_sort: Cached topological sort of all components in the pipeline.
1539 :returns:
1540 The name of the component to run and the cached topological sort of all components in the pipeline.
1541 """
1542 # Create a list of all components that have the same priority as the current component, including the
1543 # current component itself and remove them from the priority queue.
1544 components_with_same_priority = [component_name]
1545 while len(priority_queue) > 0:
1546 next_priority, next_component_name = priority_queue.peek()
1547 if priority == ComponentPriority.DEFER and next_priority == ComponentPriority.DEFER:
1548 priority_queue.pop() # actually remove the component
1549 components_with_same_priority.append(next_component_name)
1550 else:
1551 break
1553 # If there are multiple components with the same priority, we tiebreak them to decide which one to run first.
1554 if len(components_with_same_priority) > 1:
1555 topological_sort = topological_sort or self._topological_sort()
1556 components_with_same_priority = sorted(
1557 components_with_same_priority, key=lambda comp_name: (topological_sort[comp_name], comp_name.lower())
1558 )
1559 component_name = components_with_same_priority[0]
1561 return component_name, topological_sort
1563 def _find_components_blocking_pipeline(
1564 self, priority_queue: FIFOPriorityQueue, component_visits: dict[str, int], inputs: InputsType
1565 ) -> tuple[list[str], list[str]]:
1566 """
1567 Finds the components that are most likely blocking the pipeline execution.
1569 :returns:
1570 The list of component names that are most likely blocking the pipeline execution and their corresponding
1571 component types.
1572 """
1573 # 1. Go through all components in priority queue (should all be blocked at this point)
1574 comps_in_queue: list[str] = [comp_name for _, _, comp_name in priority_queue._queue]
1576 # 2. Check which components have entries in inputs.
1577 comps_with_inputs = []
1578 for comp_name in comps_in_queue:
1579 # If component has non-empty inputs and is blocked it means that the component is waiting for more inputs
1580 # to run, so it could be blocking the pipeline.
1581 if inputs.get(comp_name):
1582 comps_with_inputs.append(comp_name)
1584 # If there are no components with any inputs we fallback to checking all components in the queue.
1585 # This isn't always ideal since already executed components can also be in the queue at this point also
1586 # with blocked priority.
1587 if not comps_with_inputs:
1588 comps_with_inputs = comps_in_queue
1590 # 3. Only keep components with the lowest number of visits. Mostly needed to handle the fallback case if no
1591 # components with inputs are found.
1592 ordered_comps_with_inputs = sorted(comps_with_inputs, key=lambda x: component_visits[x])
1593 lowest_component_visit = component_visits[ordered_comps_with_inputs[0]]
1594 possible_blocking_comps = [
1595 comp for comp in ordered_comps_with_inputs if component_visits[comp] == lowest_component_visit
1596 ]
1598 # If there is only one component with the lowest visits, return it as the most likely blocking component.
1599 if len(possible_blocking_comps) == 1:
1600 blocking_comp_types = [self.graph.nodes[possible_blocking_comps[0]]["instance"].__class__.__name__]
1601 self._log_warning_for_blocking_components(
1602 blocking_comp_names=possible_blocking_comps, blocking_comp_types=blocking_comp_types
1603 )
1604 return possible_blocking_comps, blocking_comp_types
1606 # 4. Then for all components with the same lowest component visits we sort topologically before returning.
1607 topological_sort = self._topological_sort()
1608 possible_blocking_comps = sorted(
1609 possible_blocking_comps, key=lambda comp_name: (topological_sort[comp_name], comp_name.lower())
1610 )
1611 possible_blocking_comp_types = [
1612 self.graph.nodes[comp_name]["instance"].__class__.__name__ for comp_name in possible_blocking_comps
1613 ]
1615 self._log_warning_for_blocking_components(
1616 blocking_comp_names=possible_blocking_comps, blocking_comp_types=possible_blocking_comp_types
1617 )
1618 return possible_blocking_comps, possible_blocking_comp_types
1620 def _log_warning_for_blocking_components(
1621 self, blocking_comp_names: list[str], blocking_comp_types: list[dict]
1622 ) -> None:
1623 """
1624 Logs a warning about the components that are most likely blocking the pipeline execution.
1626 :param blocking_comp_names: The list of component names that are most likely blocking the pipeline execution.
1627 :param blocking_comp_types: The list of component types that are most likely blocking the pipeline execution.
1628 """
1629 comp_details = "\n".join(
1630 f" - '{name}' ({comp_type})"
1631 for name, comp_type in zip(blocking_comp_names, blocking_comp_types, strict=True)
1632 )
1633 logger.warning(
1634 "Cannot run pipeline - the pipeline appears to be blocked.\n"
1635 "The following components could not be run and may be waiting on inputs that will "
1636 "never arrive:\n" + comp_details + "\n"
1637 "Note that some of these components may be intentionally inactive due to conditional "
1638 "branching. If this is unexpected, check the connections to these components and "
1639 "ensure all required inputs are provided.",
1640 component_names=blocking_comp_names,
1641 component_types=blocking_comp_types,
1642 )
1644 def _write_component_outputs(
1645 self,
1646 *,
1647 component_name: str,
1648 component_outputs: Mapping[str, Any],
1649 inputs: InputsType,
1650 receivers: Sequence[tuple[str, OutputSocket, InputSocket, ConversionStrategyType]],
1651 include_outputs_from: set[str],
1652 ) -> Mapping[str, Any]:
1653 """
1654 Distributes the outputs of a component to the input sockets that it is connected to.
1656 :param component_name: The name of the component.
1657 :param component_outputs: The outputs of the component.
1658 :param inputs: The current global input state.
1659 :param receivers: A sequence of tuples containing:
1660 - receiver component name,
1661 - output socket of the sender,
1662 - input socket of the receiver,
1663 - ConversionStrategy to be used to convert the value if required, otherwise None.
1664 :param include_outputs_from: Set of component names that should always return an output from the pipeline.
1665 """
1666 for receiver_name, sender_socket, receiver_socket, conversion_strategy in receivers:
1667 # We either get the value that was produced by the actor or we use a _NoOutputProduced marker to indicate
1668 # that the sender did not produce an output for this socket.
1669 # This allows us to track if a predecessor already ran but did not produce an output.
1670 value = component_outputs.get(sender_socket.name, _NoOutputProduced())
1672 if not isinstance(value, _NoOutputProduced) and conversion_strategy:
1673 try:
1674 value = _convert_value(value=value, conversion_strategy=conversion_strategy)
1675 except Exception as e:
1676 sender_node = self.graph.nodes.get(component_name)
1677 sender_instance = sender_node.get("instance") if sender_node else None
1678 sender_type_name = type(sender_instance).__name__ if sender_instance else "unknown"
1680 receiver_node = self.graph.nodes.get(receiver_name)
1681 receiver_instance = receiver_node.get("instance") if receiver_node else None
1682 receiver_type_name = type(receiver_instance).__name__ if receiver_instance else "unknown"
1684 msg = (
1685 f"Failed to perform conversion between components:\n"
1686 f"Sender component: '{component_name}' (type: '{sender_type_name}')\n"
1687 f"Sender socket: '{sender_socket.name}'\n"
1688 f"Receiver component: '{receiver_name}' (type: '{receiver_type_name}')\n"
1689 f"Receiver socket: '{receiver_socket.name}'\n"
1690 f"Error: {e}"
1691 )
1692 raise PipelineRuntimeError(component_name=None, component_type=None, message=msg) from e
1694 if receiver_name not in inputs:
1695 inputs[receiver_name] = {}
1697 if receiver_socket.is_lazy_variadic:
1698 # If the receiver socket is lazy variadic, we append the new input.
1699 # Lazy variadic sockets can collect multiple inputs.
1700 _write_to_lazy_variadic_socket(
1701 inputs=inputs,
1702 receiver_name=receiver_name,
1703 receiver_socket_name=receiver_socket.name,
1704 component_name=component_name,
1705 value=value,
1706 )
1707 else:
1708 # If the receiver socket is not lazy variadic, it is greedy variadic or non-variadic.
1709 # We overwrite with the new input if it's not a _NoOutputProduced marker, or if the current value
1710 # is None.
1711 _write_to_standard_socket(
1712 inputs=inputs,
1713 receiver_name=receiver_name,
1714 receiver_socket_name=receiver_socket.name,
1715 component_name=component_name,
1716 value=value,
1717 )
1719 # If we want to include all outputs from this actor in the final outputs, we don't need to prune any consumed
1720 # outputs
1721 if component_name in include_outputs_from:
1722 return component_outputs
1724 # We prune outputs that were consumed by any receiving sockets.
1725 # All remaining outputs will be added to the final outputs of the pipeline.
1726 consumed_outputs = {sender_socket.name for _, sender_socket, __, ___ in receivers}
1727 return {key: value for key, value in component_outputs.items() if key not in consumed_outputs}
1729 @staticmethod
1730 def _is_queue_stale(priority_queue: FIFOPriorityQueue) -> bool:
1731 """
1732 Checks if the priority queue needs to be recomputed because the priorities might have changed.
1734 The queue is considered stale if it is empty or if the highest priority component is not READY or HIGHEST.
1736 For example, if the next component in a queue has the priority READY then the equality becomes
1737 ComponentPriority.READY > ComponentPriority.READY which is false.
1738 However, if the next component has priority DEFER (or BLOCKED) then the equality becomes
1739 ComponentPriority.DEFER > ComponentPriority.READY which is true, indicating that the queue is stale.
1741 :param priority_queue: Priority queue of component names.
1742 """
1743 return len(priority_queue) == 0 or priority_queue.peek()[0] > ComponentPriority.READY
1745 @staticmethod
1746 def validate_pipeline(priority_queue: FIFOPriorityQueue) -> None:
1747 """
1748 Validate the pipeline to check if it is blocked or has no valid entry point.
1750 :param priority_queue: Priority queue of component names.
1751 :raises PipelineRuntimeError:
1752 If the pipeline is blocked or has no valid entry point.
1753 """
1754 if len(priority_queue) == 0:
1755 return
1757 candidate = priority_queue.peek()
1758 if candidate is not None and candidate[0] == ComponentPriority.BLOCKED:
1759 raise PipelineComponentsBlockedError()
1761 def _find_super_components(self) -> list[tuple[str, Component]]:
1762 """
1763 Find all SuperComponents in the pipeline.
1765 :returns:
1766 List of tuples containing (component_name, component_instance) representing a SuperComponent.
1767 """
1769 super_components = []
1770 for comp_name, comp in self.walk():
1771 # a SuperComponent has a "pipeline" attribute which itself a Pipeline instance
1772 # we don't test against SuperComponent because doing so always lead to circular imports
1773 if hasattr(comp, "pipeline") and isinstance(comp.pipeline, self.__class__):
1774 super_components.append((comp_name, comp))
1775 return super_components
1777 def _merge_super_component_pipelines(self) -> tuple[networkx.MultiDiGraph, dict[str, str]]:
1778 """
1779 Merge the internal pipelines of SuperComponents into the main pipeline graph structure.
1781 This creates a new networkx.MultiDiGraph containing all the components from both the main pipeline
1782 and all the internal SuperComponents' pipelines. The SuperComponents are removed and their internal
1783 components are connected to corresponding input and output sockets of the main pipeline.
1785 :returns:
1786 A tuple containing:
1787 - A networkx.MultiDiGraph with the expanded structure of the main pipeline and all it's SuperComponents
1788 - A dictionary mapping component names to boolean indicating that this component was part of a
1789 SuperComponent
1790 - A dictionary mapping component names to their SuperComponent name
1791 """
1792 merged_graph = self.graph.copy()
1793 super_component_mapping: dict[str, str] = {}
1795 for super_name, super_component in self._find_super_components():
1796 internal_pipeline = super_component.pipeline # type: ignore
1797 internal_graph = internal_pipeline.graph.copy()
1799 # Mark all components in the internal pipeline as being part of a SuperComponent
1800 for node in internal_graph.nodes():
1801 super_component_mapping[node] = super_name
1803 # edges connected to the super component
1804 incoming_edges = list(merged_graph.in_edges(super_name, data=True))
1805 outgoing_edges = list(merged_graph.out_edges(super_name, data=True))
1807 # merge the SuperComponent graph into the main graph and remove the super component node
1808 # since its components are now part of the main graph
1809 merged_graph = networkx.compose(merged_graph, internal_graph)
1810 merged_graph.remove_node(super_name)
1812 # get the entry and exit points of the SuperComponent internal pipeline
1813 entry_points = [n for n in internal_graph.nodes() if internal_graph.in_degree(n) == 0]
1814 exit_points = [n for n in internal_graph.nodes() if internal_graph.out_degree(n) == 0]
1816 # connect the incoming edges to entry points
1817 for sender, _, edge_data in incoming_edges:
1818 sender_socket = edge_data["from_socket"]
1819 for entry_point in entry_points:
1820 # find a matching input socket in the entry point
1821 entry_point_sockets = internal_graph.nodes[entry_point]["input_sockets"]
1822 for socket_name, socket in entry_point_sockets.items():
1823 if _types_are_compatible(sender_socket.type, socket.type, self._connection_type_validation)[0]:
1824 merged_graph.add_edge(
1825 sender,
1826 entry_point,
1827 key=f"{sender_socket.name}/{socket_name}",
1828 conn_type=_type_name(sender_socket.type),
1829 from_socket=sender_socket,
1830 to_socket=socket,
1831 mandatory=socket.is_mandatory,
1832 )
1834 # connect outgoing edges from exit points
1835 for _, receiver, edge_data in outgoing_edges:
1836 receiver_socket = edge_data["to_socket"]
1837 for exit_point in exit_points:
1838 # find a matching output socket in the exit point
1839 exit_point_sockets = internal_graph.nodes[exit_point]["output_sockets"]
1840 for socket_name, socket in exit_point_sockets.items():
1841 if _types_are_compatible(socket.type, receiver_socket.type, self._connection_type_validation)[
1842 0
1843 ]:
1844 merged_graph.add_edge(
1845 exit_point,
1846 receiver,
1847 key=f"{socket_name}/{receiver_socket.name}",
1848 conn_type=_type_name(socket.type),
1849 from_socket=socket,
1850 to_socket=receiver_socket,
1851 mandatory=receiver_socket.is_mandatory,
1852 )
1854 return merged_graph, super_component_mapping
1856 def _is_pipeline_possibly_blocked(self, current_pipeline_outputs: dict[str, Any]) -> bool:
1857 """
1858 Heuristically determines whether the pipeline is possibly blocked based on its current outputs.
1860 This method checks if the pipeline has produced any of the expected outputs.
1861 - If no outputs are expected (i.e., `self.outputs()` returns an empty list), the method assumes the pipeline
1862 is not blocked.
1863 - If at least one expected output is present in `current_pipeline_outputs`, the pipeline is also assumed to not
1864 be blocked.
1865 - If none of the expected outputs are present, the pipeline is considered to be possibly blocked.
1867 Note: This check is not definitive—it is intended as a best-effort guess to detect a stalled or misconfigured
1868 pipeline when there are no more runnable components.
1870 :param current_pipeline_outputs: A dictionary of outputs currently produced by the pipeline.
1871 :returns:
1872 bool: True if the pipeline is possibly blocked (i.e., expected outputs are missing), False otherwise.
1873 """
1874 expected_outputs = self.outputs()
1875 return bool(expected_outputs) and not any(k in current_pipeline_outputs for k in expected_outputs)
1878def _connections_status(
1879 sender_node: str, receiver_node: str, sender_sockets: list[OutputSocket], receiver_sockets: list[InputSocket]
1880) -> str:
1881 """
1882 Lists the status of the sockets, for error messages.
1883 """
1884 sender_sockets_entries = []
1885 for sender_socket in sender_sockets:
1886 sender_sockets_entries.append(f" - {sender_socket.name}: {_type_name(sender_socket.type)}")
1887 sender_sockets_list = "\n".join(sender_sockets_entries)
1889 receiver_sockets_entries = []
1890 for receiver_socket in receiver_sockets:
1891 if receiver_socket.senders:
1892 sender_status = f"sent by {','.join(receiver_socket.senders)}"
1893 else:
1894 sender_status = "available"
1895 receiver_sockets_entries.append(
1896 f" - {receiver_socket.name}: {_type_name(receiver_socket.type)} ({sender_status})"
1897 )
1898 receiver_sockets_list = "\n".join(receiver_sockets_entries)
1900 return f"'{sender_node}':\n{sender_sockets_list}\n'{receiver_node}':\n{receiver_sockets_list}"
1903# Utility functions
1906def _validate_component_output_keys(
1907 component_name: str, comp: dict[str, Any], component_output: Mapping[str, Any]
1908) -> None:
1909 """
1910 Validate that the output keys returned by a component match its declared output types.
1912 Logs a warning for any actually returned output key(s) that was not declared as an output socket(s).
1913 This helps catch bugs where a component returns wrong keys, which would otherwise cause downstream components to
1914 wait forever for expected data, resulting in a confusing "Pipeline Blocked" error that points to an unexpected
1915 component.
1917 :param component_name: Name of the Component as registered in the Pipeline.
1918 :param comp: The component metadata dictionary containing the component instance and its input/output socket
1919 metadata.
1920 :param component_output: The actual output dictionary returned by the component's run method.
1921 """
1922 output_sockets = comp.get("output_sockets", {})
1923 if not output_sockets:
1924 return
1926 declared_keys = set(output_sockets.keys())
1927 actual_keys = set(component_output.keys())
1929 extra_keys = actual_keys - declared_keys
1931 instance = comp["instance"]
1932 component_type = instance.__class__.__name__
1934 if extra_keys:
1935 logger.warning(
1936 "Component '{component_name}' (type: {component_type}) returned output keys {extra_keys} "
1937 "that are not declared in its output types. "
1938 "These keys will be ignored and not passed to downstream components. "
1939 "Make sure the component's output keys match its declared @component.output_types.",
1940 component_name=component_name,
1941 component_type=component_type,
1942 extra_keys=extra_keys,
1943 )
1946def _write_to_lazy_variadic_socket(
1947 inputs: InputsType, receiver_name: str, receiver_socket_name: str, component_name: str, value: Any
1948) -> None:
1949 """
1950 Write to a lazy variadic socket.
1952 Mutates inputs in place.
1954 :param inputs: The global inputs state to be mutated.
1955 :param receiver_name: The name of the component receiving the input.
1956 :param receiver_socket_name: The name of the socket receiving the input.
1957 :param component_name: The name of the component sending the input.
1958 :param value: The value to be sent to the socket.
1959 """
1960 if not inputs[receiver_name].get(receiver_socket_name):
1961 inputs[receiver_name][receiver_socket_name] = []
1963 inputs[receiver_name][receiver_socket_name].append({"sender": component_name, "value": value})
1966def _write_to_standard_socket(
1967 inputs: InputsType, receiver_name: str, receiver_socket_name: str, component_name: str, value: Any
1968) -> None:
1969 """
1970 Write to a greedy variadic or non-variadic socket.
1972 Mutates inputs in place.
1974 :param inputs: The global inputs state to be mutated.
1975 :param receiver_name: The name of the component receiving the input.
1976 :param receiver_socket_name: The name of the socket receiving the input.
1977 :param component_name: The name of the component sending the input.
1978 :param value: The value to be sent to the socket.
1979 """
1980 current_value = inputs[receiver_name].get(receiver_socket_name)
1982 # Only overwrite if there's no existing value, or we have a new value to provide
1983 if current_value is None or not isinstance(value, _NoOutputProduced):
1984 inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}]