Coverage for haystack/core/pipeline/base.py: 92%
640 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-10-09 14:34 +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 edge_key = f"{sender_socket.name}/{receiver_socket.name}"
768 if self.graph.has_edge(sender_component_name, receiver_component_name, key=edge_key):
769 # This is already connected, nothing to do
770 return self
772 if receiver_socket.senders:
773 receiver_socket = self._make_socket_auto_variadic(
774 component_name=receiver_component_name, receiver_socket=receiver_socket, error_type=PipelineConnectError
775 )
777 # Update the sockets with the new connection
778 sender_socket.receivers.append(receiver_component_name)
779 receiver_socket.senders.append(sender_component_name)
781 # Create the new connection
782 self.graph.add_edge(
783 sender_component_name,
784 receiver_component_name,
785 key=f"{sender_socket.name}/{receiver_socket.name}",
786 conn_type=_type_name(sender_socket.type),
787 from_socket=sender_socket,
788 to_socket=receiver_socket,
789 mandatory=receiver_socket.is_mandatory,
790 conversion_strategy=conversion_strategy,
791 )
792 return self
794 def connect_many(self, connections: list[tuple[str, str]]) -> Self:
795 """
796 Connect multiple pairs of components.
798 Connections are made in the order provided. If connecting a pair raises an exception, no subsequent pairs
799 are connected, while earlier successful connections remain in the pipeline. Repeating an existing connection
800 is a no-op.
802 :param connections:
803 A list of `(sender, receiver)` pairs. Each value uses the same format accepted by
804 `Pipeline.connect()`.
805 :returns:
806 The Pipeline instance.
808 :raises PipelineConnectError:
809 If a pair of components cannot be connected.
810 :raises ValueError:
811 If a sender or receiver component is not present in the pipeline.
812 """
813 for sender, receiver in connections:
814 self.connect(sender, receiver)
816 return self
818 def get_component(self, name: str) -> Component:
819 """
820 Get the component with the specified name from the pipeline.
822 :param name:
823 The name of the component.
824 :returns:
825 The instance of that component.
827 :raises ValueError:
828 If a component with that name is not present in the pipeline.
829 """
830 try:
831 return self.graph.nodes[name]["instance"]
832 except KeyError as exc:
833 raise ValueError(f"Component named {name} not found in the pipeline.") from exc
835 def get_component_name(self, instance: Component) -> str:
836 """
837 Returns the name of the Component instance if it has been added to this Pipeline or an empty string otherwise.
839 :param instance:
840 The Component instance to look for.
841 :returns:
842 The name of the Component instance.
843 """
844 for name, inst in self.graph.nodes(data="instance"):
845 if inst == instance:
846 return name
847 return ""
849 def inputs(self, include_components_with_connected_inputs: bool = False) -> dict[str, dict[str, Any]]:
850 """
851 Returns a dictionary containing the inputs of a pipeline.
853 Each key in the dictionary corresponds to a component name, and its value is another dictionary that describes
854 the input sockets of that component, including their types and whether they are optional.
856 :param include_components_with_connected_inputs:
857 If `False`, only components that have disconnected input edges are
858 included in the output.
859 :returns:
860 A dictionary where each key is a pipeline component name and each value is a dictionary of
861 inputs sockets of that component.
862 """
863 inputs: dict[str, dict[str, Any]] = {}
864 for component_name, data in find_pipeline_inputs(self.graph, include_components_with_connected_inputs).items():
865 sockets_description = {}
866 for socket in data:
867 # Variadic mandatory sockets with existing connections don't require user input, so treat them as
868 # optional.
869 is_mandatory = socket.is_mandatory and not socket.senders
870 sockets_description[socket.name] = {"type": socket.type, "is_mandatory": is_mandatory}
871 if not socket.is_mandatory:
872 sockets_description[socket.name]["default_value"] = socket.default_value
874 if sockets_description:
875 inputs[component_name] = sockets_description
876 return inputs
878 def outputs(self, include_components_with_connected_outputs: bool = False) -> dict[str, dict[str, Any]]:
879 """
880 Returns a dictionary containing the outputs of a pipeline.
882 Each key in the dictionary corresponds to a component name, and its value is another dictionary that describes
883 the output sockets of that component.
885 :param include_components_with_connected_outputs:
886 If `False`, only components that have disconnected output edges are
887 included in the output.
888 :returns:
889 A dictionary where each key is a pipeline component name and each value is a dictionary of
890 output sockets of that component.
891 """
892 return {
893 comp: {socket.name: {"type": socket.type} for socket in data}
894 for comp, data in find_pipeline_outputs(self.graph, include_components_with_connected_outputs).items()
895 if data
896 }
898 def show(
899 self,
900 *,
901 server_url: str = "https://mermaid.ink",
902 params: dict | None = None,
903 timeout: int = 30,
904 super_component_expansion: bool = False,
905 ) -> None:
906 """
907 Display an image representing this `Pipeline` in a Jupyter notebook.
909 This function generates a diagram of the `Pipeline` using a Mermaid server and displays it directly in
910 the notebook.
912 :param server_url:
913 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
914 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
915 info on how to set up your own Mermaid server.
917 :param params:
918 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
919 Supported keys:
920 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
921 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
922 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
923 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
924 - width: Width of the output image (integer).
925 - height: Height of the output image (integer).
926 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
927 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
928 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
929 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
931 :param timeout:
932 Timeout in seconds for the request to the Mermaid server.
934 :param super_component_expansion:
935 If set to True and the pipeline contains SuperComponents the diagram will show the internal structure of
936 super-components as if they were components part of the pipeline instead of a "black-box".
937 Otherwise, only the super-component itself will be displayed.
939 :raises PipelineDrawingError:
940 If the function is called outside of a Jupyter notebook or if there is an issue with rendering.
941 """
943 if is_in_jupyter():
944 from IPython.display import Image, display
946 if super_component_expansion:
947 graph, super_component_mapping = self._merge_super_component_pipelines()
948 else:
949 graph = self.graph
950 super_component_mapping = None
952 image_data = _to_mermaid_image(
953 graph,
954 server_url=server_url,
955 params=params,
956 timeout=timeout,
957 super_component_mapping=super_component_mapping,
958 )
959 display(Image(image_data))
960 else:
961 msg = "This method is only supported in Jupyter notebooks. Use Pipeline.draw() to save an image locally."
962 raise PipelineDrawingError(msg)
964 def draw(
965 self,
966 *,
967 path: Path,
968 server_url: str = "https://mermaid.ink",
969 params: dict | None = None,
970 timeout: int = 30,
971 super_component_expansion: bool = False,
972 ) -> None:
973 """
974 Save an image representing this `Pipeline` to the specified file path.
976 This function generates a diagram of the `Pipeline` using the Mermaid server and saves it to the provided path.
978 :param path:
979 The file path where the generated image will be saved.
981 :param server_url:
982 The base URL of the Mermaid server used for rendering (default: 'https://mermaid.ink').
983 See https://github.com/jihchi/mermaid.ink and https://github.com/mermaid-js/mermaid-live-editor for more
984 info on how to set up your own Mermaid server.
986 :param params:
987 Dictionary of customization parameters to modify the output. Refer to Mermaid documentation for more details
988 Supported keys:
989 - format: Output format ('img', 'svg', or 'pdf'). Default: 'img'.
990 - type: Image type for /img endpoint ('jpeg', 'png', 'webp'). Default: 'png'.
991 - theme: Mermaid theme ('default', 'neutral', 'dark', 'forest'). Default: 'neutral'.
992 - bgColor: Background color in hexadecimal (e.g., 'FFFFFF') or named format (e.g., '!white').
993 - width: Width of the output image (integer).
994 - height: Height of the output image (integer).
995 - scale: Scaling factor (1–3). Only applicable if 'width' or 'height' is specified.
996 - fit: Whether to fit the diagram size to the page (PDF only, boolean).
997 - paper: Paper size for PDFs (e.g., 'a4', 'a3'). Ignored if 'fit' is true.
998 - landscape: Landscape orientation for PDFs (boolean). Ignored if 'fit' is true.
1000 :param timeout:
1001 Timeout in seconds for the request to the Mermaid server.
1003 :param super_component_expansion:
1004 If set to True and the pipeline contains SuperComponents the diagram will show the internal structure of
1005 super-components as if they were components part of the pipeline instead of a "black-box".
1006 Otherwise, only the super-component itself will be displayed.
1008 :raises PipelineDrawingError:
1009 If there is an issue with rendering or saving the image.
1010 """
1012 # Before drawing we edit a bit the graph, to avoid modifying the original that is
1013 # used for running the pipeline we copy it.
1014 if super_component_expansion:
1015 graph, super_component_mapping = self._merge_super_component_pipelines()
1016 else:
1017 graph = self.graph
1018 super_component_mapping = None
1020 image_data = _to_mermaid_image(
1021 graph,
1022 server_url=server_url,
1023 params=params,
1024 timeout=timeout,
1025 super_component_mapping=super_component_mapping,
1026 )
1027 Path(path).write_bytes(image_data)
1029 def walk(self) -> Iterator[tuple[str, Component]]:
1030 """
1031 Visits each component in the pipeline exactly once and yields its name and instance.
1033 No guarantees are provided on the visiting order.
1035 :returns:
1036 An iterator of tuples of component name and component instance.
1037 """
1038 for component_name, instance in self.graph.nodes(data="instance"): # noqa: UP028
1039 yield component_name, instance
1041 def warm_up(self) -> None:
1042 """
1043 Make sure all components are warm.
1045 It's the component's responsibility to make sure this method can be called at every `Pipeline.run()`
1046 without re-initializing everything.
1047 """
1048 for component_name in self.graph.nodes:
1049 if hasattr(self.graph.nodes[component_name]["instance"], "warm_up"):
1050 logger.info("Warming up component {component_name}...", component_name=component_name)
1051 self.graph.nodes[component_name]["instance"].warm_up()
1053 async def warm_up_async(self) -> None:
1054 """
1055 Make sure all components are warm, using the async warm-up path where available.
1057 Each component is warmed up with `warm_up_async` if it has one, otherwise with its sync `warm_up`.
1058 Both run on the event loop, never offloaded to a worker thread.
1059 This ensures that if an async client is created during `warm-up` (residual scenario), it binds to the loop that
1060 `run_async` will use.
1061 """
1062 for component_name in self.graph.nodes:
1063 instance = self.graph.nodes[component_name]["instance"]
1064 if hasattr(instance, "warm_up_async"):
1065 logger.info("Warming up component {component_name}...", component_name=component_name)
1066 await instance.warm_up_async()
1067 elif hasattr(instance, "warm_up"):
1068 logger.info("Warming up component {component_name}...", component_name=component_name)
1069 instance.warm_up()
1071 def close(self) -> None:
1072 """
1073 Release resources held by the pipeline's components by calling each component's `close` method.
1075 Only the synchronous side of each component is released here; use `close_async` to release async clients.
1076 """
1077 for component_name in self.graph.nodes:
1078 instance = self.graph.nodes[component_name]["instance"]
1079 if hasattr(instance, "close"):
1080 logger.info("Closing component {component_name}...", component_name=component_name)
1081 instance.close()
1083 async def close_async(self) -> None:
1084 """
1085 Release resources held by the pipeline's components, using the async close path where available.
1087 Each component is closed with `close_async` if it has one, otherwise with its sync `close`.
1088 """
1089 for component_name in self.graph.nodes:
1090 instance = self.graph.nodes[component_name]["instance"]
1091 if hasattr(instance, "close_async"):
1092 logger.info("Closing component {component_name}...", component_name=component_name)
1093 await instance.close_async()
1094 elif hasattr(instance, "close"):
1095 logger.info("Closing component {component_name}...", component_name=component_name)
1096 instance.close()
1098 @staticmethod
1099 def _create_component_span(
1100 component_name: str, instance: Component, inputs: dict[str, Any], parent_span: tracing.Span | None = None
1101 ) -> ContextManager[tracing.Span]:
1102 return tracing.tracer.trace(
1103 "haystack.component.run",
1104 tags={
1105 "haystack.component.name": component_name,
1106 "haystack.component.type": instance.__class__.__name__,
1107 "haystack.component.fully_qualified_type": generate_qualified_class_name(type(instance)),
1108 "haystack.component.input_types": {k: type(v).__name__ for k, v in inputs.items()},
1109 "haystack.component.input_spec": {
1110 key: {
1111 "type": value.type.__name__ if type(value.type) is type else str(value.type),
1112 "senders": value.senders,
1113 }
1114 for key, value in instance.__haystack_input__._sockets_dict.items() # type: ignore
1115 },
1116 "haystack.component.output_spec": {
1117 key: {
1118 "type": value.type.__name__ if type(value.type) is type else str(value.type),
1119 "receivers": value.receivers,
1120 }
1121 for key, value in instance.__haystack_output__._sockets_dict.items() # type: ignore
1122 },
1123 },
1124 parent_span=parent_span,
1125 )
1127 def validate_input(self, data: dict[str, Any]) -> None:
1128 """
1129 Validates pipeline input data.
1131 Validates that data:
1132 * Each Component name actually exists in the Pipeline
1133 * Each Component is not missing any input
1134 * Each Component has only one input per input socket, if not variadic
1135 * Each Component doesn't receive inputs that are already sent by another Component
1137 :param data:
1138 A dictionary of inputs for the pipeline's components. Each key is a component name.
1140 :raises ValueError:
1141 If inputs are invalid according to the above.
1142 """
1143 for component_name, component_inputs in data.items():
1144 # Check that the component exists
1145 if component_name not in self.graph.nodes:
1146 raise ValueError(f"Component named '{component_name}' not found in the pipeline.")
1147 # Check that no input is provided that the component can't accept
1148 instance = self.graph.nodes[component_name]["instance"]
1149 for input_name in component_inputs.keys():
1150 if input_name not in instance.__haystack_input__._sockets_dict:
1151 raise ValueError(f"Input '{input_name}' not found in component '{component_name}'.")
1153 for component_name in self.graph.nodes:
1154 instance = self.graph.nodes[component_name]["instance"]
1155 for socket_name, socket in instance.__haystack_input__._sockets_dict.items():
1156 component_inputs = data.get(component_name, {})
1157 # Check that no mandatory input is missing for any component in the pipeline
1158 if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
1159 raise ValueError(f"Missing mandatory input '{socket_name}' for component '{component_name}'.")
1161 # Check if an input is provided more than once for non-variadic sockets
1162 if socket.senders and socket_name in component_inputs:
1163 self._make_socket_auto_variadic(
1164 component_name=component_name, receiver_socket=socket, error_type=ValueError
1165 )
1167 def _make_socket_auto_variadic(
1168 self, component_name: str, receiver_socket: InputSocket, error_type: type[Exception]
1169 ) -> InputSocket:
1170 """
1171 Attempts to make the receiver socket lazy variadic in-place to accommodate a new sender.
1173 A socket is automatically made lazy variadic when:
1174 - It already has at least one connected sender
1175 - It is not already variadic
1176 - Its type is list, Optional[list], a union of list types
1178 When auto-variadicity is applied, `wrap_input_in_list` is also set to False so that sender output types match
1179 the receiver socket's declared list type directly.
1181 :param component_name:
1182 Name of the component owning the receiver socket, used in error messages.
1183 :param receiver_socket:
1184 The receiver socket to inspect and potentially modify in-place.
1185 :param error_type:
1186 Exception class to raise on failure (e.g. ValueError or PipelineConnectError).
1187 :returns:
1188 The (possibly modified) receiver socket.
1189 :raises error_type:
1190 If the socket cannot accept multiple senders given its type constraints.
1191 """
1192 # If it's already variadic, we return as-is
1193 if receiver_socket.is_variadic:
1194 return receiver_socket
1196 # Get receiver origin
1197 receiver_origin = _safe_get_origin(receiver_socket.type)
1199 # Handle Union types
1200 if receiver_origin == Union:
1201 # Unwrap Optional types
1202 non_none_args = [a for a in get_args(receiver_socket.type) if a is not type(None)]
1203 if len(non_none_args) == 1:
1204 receiver_origin = _safe_get_origin(non_none_args[0])
1205 # Handle Union of list types (e.g. list[int] | list[str])
1206 elif all(_safe_get_origin(arg) == list for arg in non_none_args):
1207 receiver_origin = list
1209 # If the receiver origin is a list, we can make the socket lazy variadic
1210 if receiver_origin == list:
1211 receiver_socket.is_lazy_variadic = True
1212 receiver_socket.wrap_input_in_list = False
1213 return receiver_socket
1215 raise error_type(
1216 f"Component '{component_name}' cannot accept multiple inputs to '{receiver_socket.name}'. "
1217 f"It is already connected to component '{receiver_socket.senders[0]}', and it can only accept "
1218 f"inputs from multiple senders if its type is list, Optional[list], or union of list types."
1219 )
1221 def _prepare_component_input_data(self, data: dict[str, Any]) -> dict[str, dict[str, Any]]:
1222 """
1223 Prepares input data for pipeline components.
1225 Organizes input data for pipeline components and identifies any inputs that are not matched to any
1226 component's input slots. Deep-copies data items to avoid sharing mutables across multiple components.
1228 This method processes a flat dictionary of input data, where each key-value pair represents an input name
1229 and its corresponding value. It distributes these inputs to the appropriate pipeline components based on
1230 their input requirements. Inputs that don't match any component's input slots are classified as unresolved.
1232 :param data:
1233 A dictionary potentially having input names as keys and input values as values.
1235 :returns:
1236 A dictionary mapping component names to their respective matched inputs.
1237 """
1238 # check whether the data is a nested dictionary of component inputs where each key is a component name
1239 # and each value is a dictionary of input parameters for that component
1240 is_nested_component_input = all(isinstance(value, dict) for value in data.values())
1241 if data and is_nested_component_input and not any(name in self.graph.nodes for name in data):
1242 # Dictionary values can also be flat inputs. Include connected sockets so the input format does not
1243 # depend on the value type, but require every key to name a socket to preserve unknown-component errors.
1244 socket_names = {
1245 socket_name
1246 for sockets in self.inputs(include_components_with_connected_inputs=True).values()
1247 for socket_name in sockets
1248 }
1249 is_nested_component_input = not data.keys() <= socket_names
1250 if not is_nested_component_input:
1251 # flat input, a dict where keys are input names and values are the corresponding values
1252 # we need to convert it to a nested dictionary of component inputs and then run the pipeline
1253 # just like in the previous case
1254 pipeline_input_data: dict[str, dict[str, Any]] = defaultdict(dict)
1255 unresolved_kwargs = {}
1257 # Retrieve the input slots for each component in the pipeline
1258 available_inputs: dict[str, dict[str, Any]] = self.inputs()
1260 # Go through all provided to distribute them to the appropriate component inputs
1261 for input_name, input_value in data.items():
1262 resolved_at_least_once = False
1264 # Check each component to see if it has a slot for the current kwarg
1265 for component_name, component_inputs in available_inputs.items():
1266 if input_name in component_inputs:
1267 # If a match is found, add the kwarg to the component's input data
1268 pipeline_input_data[component_name][input_name] = input_value
1269 resolved_at_least_once = True
1271 if not resolved_at_least_once:
1272 unresolved_kwargs[input_name] = input_value
1274 if unresolved_kwargs:
1275 logger.warning(
1276 "Inputs {input_keys} were not matched to any component inputs, please check your run parameters.",
1277 input_keys=list(unresolved_kwargs.keys()),
1278 )
1280 data = dict(pipeline_input_data)
1282 # deepcopying the inputs prevents the Pipeline run logic from being altered unexpectedly
1283 # when the same input reference is passed to multiple components.
1284 for component_name, component_inputs in data.items():
1285 data[component_name] = {k: _deepcopy_with_exceptions(v) for k, v in component_inputs.items()}
1287 return data
1289 def _find_receivers_from(
1290 self, component_name: str
1291 ) -> list[tuple[str, OutputSocket, InputSocket, ConversionStrategyType]]:
1292 """
1293 Utility function to find all Components that receive input from `component_name`.
1295 :param component_name:
1296 Name of the sender Component
1298 :returns: A list of tuples containing:
1299 - receiver component name
1300 - sender OutputSocket
1301 - receiver InputSocket
1302 - ConversionStrategy if conversion is required, otherwise None.
1303 """
1304 res = []
1305 for _, receiver_name, connection in self.graph.edges(nbunch=component_name, data=True):
1306 sender_socket: OutputSocket = connection["from_socket"]
1307 receiver_socket: InputSocket = connection["to_socket"]
1308 res.append((receiver_name, sender_socket, receiver_socket, connection.get("conversion_strategy")))
1309 return res
1311 @staticmethod
1312 def _convert_to_internal_format(pipeline_inputs: dict[str, Any]) -> InputsType:
1313 """
1314 Converts the inputs to the pipeline to the format that is needed for the internal `Pipeline.run` logic.
1316 Example Input:
1317 {'prompt_builder': {'question': 'Who lives in Paris?'}, 'retriever': {'query': 'Who lives in Paris?'}}
1318 Example Output:
1319 {'prompt_builder': {'question': [{'sender': None, 'value': 'Who lives in Paris?'}]},
1320 'retriever': {'query': [{'sender': None, 'value': 'Who lives in Paris?'}]}}
1322 :param pipeline_inputs: Inputs to the pipeline.
1323 :returns: Converted inputs that can be used by the internal `Pipeline.run` logic.
1324 """
1325 inputs: InputsType = {}
1326 for component_name, socket_dict in pipeline_inputs.items():
1327 inputs[component_name] = {}
1328 for socket_name, value in socket_dict.items():
1329 inputs[component_name][socket_name] = [{"sender": None, "value": value}]
1331 return inputs
1333 @staticmethod
1334 def _consume_component_inputs(
1335 component_name: str, component: dict, inputs: InputsType, is_resume: bool = False
1336 ) -> dict[str, Any]:
1337 """
1338 Extracts the inputs needed to run for the component and removes them from the global inputs state.
1340 :param component_name: The name of a component.
1341 :param component: Component with component metadata.
1342 :param inputs: Global inputs state.
1343 :param is_resume: Whether the component is being resumed from a breakpoint. If True, the inputs
1344 have already been consumed, so the first available value is returned for each socket.
1345 :returns: The inputs for the component.
1346 """
1347 component_inputs = inputs.get(component_name, {})
1348 consumed_inputs = {}
1349 greedy_inputs_to_remove = set()
1350 for socket_name, socket in component["input_sockets"].items():
1351 socket_inputs = component_inputs.get(socket_name, [])
1352 socket_inputs_values = [
1353 sock["value"] for sock in socket_inputs if not isinstance(sock["value"], _NoOutputProduced)
1354 ]
1356 # if we are resuming a component, the inputs are already consumed, so we just return the first input
1357 if is_resume:
1358 consumed_inputs[socket_name] = socket_inputs_values[0]
1359 continue
1361 if socket_inputs_values:
1362 if socket.is_greedy:
1363 # We need to keep track of greedy inputs because we always remove them, even if they come from
1364 # outside the pipeline. Otherwise, a greedy input from the user would trigger a pipeline to run
1365 # indefinitely.
1366 greedy_inputs_to_remove.add(socket_name)
1367 consumed_inputs[socket_name] = [socket_inputs_values[0]]
1368 elif socket.is_lazy_variadic:
1369 if socket.wrap_input_in_list:
1370 # We use all inputs provided to the socket on a lazy variadic socket.
1371 # So keep it wrapped in a list.
1372 consumed_inputs[socket_name] = socket_inputs_values
1373 else:
1374 # We flatten one-level of lists for lazy variadic sockets that don't wrap inputs in lists.
1375 # This way the incoming inputs match the expected type of the socket.
1376 consumed_inputs[socket_name] = list(itertools.chain.from_iterable(socket_inputs_values))
1377 else:
1378 # For a normal socket we only care about the first input provided to the socket.
1379 consumed_inputs[socket_name] = socket_inputs_values[0]
1381 # We prune all inputs except for those that were provided from outside the pipeline (e.g. user inputs).
1382 pruned_inputs = {
1383 socket_name: [
1384 sock for sock in socket if sock["sender"] is None and socket_name not in greedy_inputs_to_remove
1385 ]
1386 for socket_name, socket in component_inputs.items()
1387 }
1388 pruned_inputs = {socket_name: socket for socket_name, socket in pruned_inputs.items() if len(socket) > 0}
1390 inputs[component_name] = pruned_inputs
1392 return consumed_inputs
1394 def _fill_queue(
1395 self, component_names: list[str], inputs: InputsType, component_visits: dict[str, int]
1396 ) -> FIFOPriorityQueue:
1397 """
1398 Calculates the execution priority for each component and inserts it into the priority queue.
1400 :param component_names: Names of the components to put into the queue.
1401 :param inputs: Inputs to the components.
1402 :param component_visits: Current state of component visits.
1403 :returns: A prioritized queue of component names.
1404 """
1405 priority_queue = FIFOPriorityQueue()
1406 for component_name in component_names:
1407 comp = self._get_component_with_graph_metadata_and_visits(component_name, component_visits[component_name])
1408 priority = self._calculate_priority(comp, inputs.get(component_name, {}))
1409 priority_queue.push(component_name, priority)
1410 return priority_queue
1412 @staticmethod
1413 def _calculate_priority(comp: dict, comp_inputs: dict[str, list[dict[str, Any]]]) -> ComponentPriority:
1414 """
1415 Calculates the execution priority for a component depending on the component's inputs.
1417 :param comp: Component metadata and component instance.
1418 :param comp_inputs: Inputs to the component.
1419 :returns: Priority value for the component.
1420 """
1421 # NOTE: Even if a component can run, it doesn't mean it's ready to run since it could be waiting for optional
1422 # inputs. This is why it's only used to determine if a component is BLOCKED or not.
1423 if not can_component_run(comp, comp_inputs):
1424 return ComponentPriority.BLOCKED
1425 if is_any_greedy_socket_ready(comp, comp_inputs) and are_all_sockets_ready(comp, comp_inputs):
1426 # This priority is explicitly used in the async run path + implicitly in _is_queue_stale
1427 # Implicit b/c it checks via ">" operator if there is a component with HIGHEST priority
1428 return ComponentPriority.HIGHEST
1429 if all_predecessors_executed(comp, comp_inputs):
1430 # This priority is explicitly used in the async run path + in _is_queue_stale
1431 return ComponentPriority.READY
1432 # If we make it here it means the component can run but is waiting for more inputs, so we give it the lowest
1433 # priority. This way, components that are ready to run will be prioritized over ones assigned with this prio.
1434 return ComponentPriority.DEFER
1436 def _get_component_with_graph_metadata_and_visits(self, component_name: str, visits: int) -> dict[str, Any]:
1437 """
1438 Returns the component instance alongside input/output-socket metadata from the graph and adds current visits.
1440 We can't store visits in the pipeline graph because this would prevent reentrance / thread-safe execution.
1442 :param component_name: The name of the component.
1443 :param visits: Number of visits for the component.
1444 :returns: dict with keys:
1445 - instance: the component instance
1446 - input_sockets: the component input sockets metadata from the graph
1447 - output_sockets: the component output sockets metadata from the graph
1448 """
1449 comp_dict = self.graph.nodes[component_name]
1450 # We inject the visits into the component dict here to avoid storing it in the graph, which would prevent
1451 # thread-safe execution
1452 return {**comp_dict, "visits": visits}
1454 def _get_next_runnable_component(
1455 self, priority_queue: FIFOPriorityQueue, component_visits: dict[str, int]
1456 ) -> tuple[ComponentPriority, str, dict[str, Any]] | None:
1457 """
1458 Returns the next runnable component alongside its metadata from the priority queue.
1460 :param priority_queue: Priority queue of component names.
1461 :param component_visits: Current state of component visits.
1462 :returns: The next runnable component, the component name, and its priority
1463 or None if no component in the queue can run.
1464 :raises: PipelineMaxComponentRuns if the next runnable component has exceeded the maximum number of runs.
1465 """
1466 item = priority_queue.get()
1468 # If no component is runnable, return None
1469 if item is None:
1470 return None
1472 component_name = item[1]
1473 comp = self._get_component_with_graph_metadata_and_visits(component_name, component_visits[component_name])
1474 # Only raise the max run count error if the component is not blocked, since if it's blocked it means it
1475 # can't run anyway.
1476 if item[0] < ComponentPriority.BLOCKED and comp["visits"] >= self._max_runs_per_component:
1477 msg = f"Maximum run count {self._max_runs_per_component} reached for component '{component_name}'"
1478 raise PipelineMaxComponentRuns(msg)
1479 return ComponentPriority(item[0]), component_name, comp
1481 @staticmethod
1482 def _add_missing_input_defaults(
1483 component_inputs: dict[str, list[dict[str, Any]]], component_input_sockets: dict[str, InputSocket]
1484 ) -> dict[str, Any]:
1485 """
1486 Updates the inputs with the default values for the inputs that are missing
1488 :param component_inputs: Inputs for the component.
1489 :param component_input_sockets: Input sockets of the component.
1490 """
1491 for name, socket in component_input_sockets.items():
1492 if not socket.is_mandatory and name not in component_inputs:
1493 # NOTE: Variadic inputs expect a single default value in the function signature that matches the inner
1494 # type, for example Variadic[str] = "default". When executed inside a pipeline, we wrap this
1495 # default into a list, resulting in ["default"], which is the intended behavior.
1496 #
1497 # However, when the component is executed directly by calling run(), the default is not wrapped and
1498 # is treated as an iterable.
1499 # For strings, this would produce ["d", "e", "f", ...] instead of ["default"].
1500 if socket.is_variadic:
1501 component_inputs[name] = [socket.default_value]
1502 else:
1503 component_inputs[name] = socket.default_value
1505 return component_inputs
1507 def _topological_sort(self) -> dict[str, int]:
1508 """
1509 Returns a topological sort of the components in the pipeline.
1511 If the graph is a DAG, we use lexicographical topological sort to get a deterministic order of the components.
1512 If the graph is not a DAG, we use the condensation of the graph to get a topological sort of the strongly
1513 connected components. This way, components that are part of the same cycle will have the same priority and will
1514 be tie-broken by their name in lexicographical order, while components that are not part of the same cycle will
1515 be tie-broken by their topological order.
1517 :returns:
1518 A dictionary mapping component names to their position in the topological sort.
1519 """
1520 if networkx.is_directed_acyclic_graph(self.graph):
1521 topological_sort = networkx.lexicographical_topological_sort(self.graph)
1522 return {node: idx for idx, node in enumerate(topological_sort)}
1523 else:
1524 condensed = networkx.condensation(self.graph)
1525 condensed_sorted = {node: idx for idx, node in enumerate(networkx.topological_sort(condensed))}
1526 return {
1527 component_name: condensed_sorted[node] for component_name, node in condensed.graph["mapping"].items()
1528 }
1530 def _tiebreak_waiting_components(
1531 self,
1532 component_name: str,
1533 priority: ComponentPriority,
1534 priority_queue: FIFOPriorityQueue,
1535 topological_sort: dict[str, int] | None,
1536 ) -> tuple[str, dict[str, int] | None]:
1537 """
1538 Decides which component to run when multiple components are waiting for inputs with the same priority.
1540 NOTE: This was designed to only tie-break for components with the priority DEFER. Since this function also
1541 removes these components from the priority queue we rely on _is_queue_stale to then refill the priority queue.
1542 And _is_queue_stale only triggers when all remaining components have BLOCKED priority.
1544 :param component_name: The name of the component.
1545 :param priority: Priority of the component.
1546 :param priority_queue: Priority queue of component names.
1547 :param topological_sort: Cached topological sort of all components in the pipeline.
1549 :returns:
1550 The name of the component to run and the cached topological sort of all components in the pipeline.
1551 """
1552 # Create a list of all components that have the same priority as the current component, including the
1553 # current component itself and remove them from the priority queue.
1554 components_with_same_priority = [component_name]
1555 while len(priority_queue) > 0:
1556 next_priority, next_component_name = priority_queue.peek()
1557 if priority == ComponentPriority.DEFER and next_priority == ComponentPriority.DEFER:
1558 priority_queue.pop() # actually remove the component
1559 components_with_same_priority.append(next_component_name)
1560 else:
1561 break
1563 # If there are multiple components with the same priority, we tiebreak them to decide which one to run first.
1564 if len(components_with_same_priority) > 1:
1565 topological_sort = topological_sort or self._topological_sort()
1566 components_with_same_priority = sorted(
1567 components_with_same_priority, key=lambda comp_name: (topological_sort[comp_name], comp_name.lower())
1568 )
1569 component_name = components_with_same_priority[0]
1571 return component_name, topological_sort
1573 def _find_components_blocking_pipeline(
1574 self, priority_queue: FIFOPriorityQueue, component_visits: dict[str, int], inputs: InputsType
1575 ) -> tuple[list[str], list[str]]:
1576 """
1577 Finds the components that are most likely blocking the pipeline execution.
1579 :returns:
1580 The list of component names that are most likely blocking the pipeline execution and their corresponding
1581 component types.
1582 """
1583 # 1. Go through all components in priority queue (should all be blocked at this point)
1584 comps_in_queue: list[str] = [comp_name for _, _, comp_name in priority_queue._queue]
1586 # 2. Check which components have entries in inputs.
1587 comps_with_inputs = []
1588 for comp_name in comps_in_queue:
1589 # If component has non-empty inputs and is blocked it means that the component is waiting for more inputs
1590 # to run, so it could be blocking the pipeline.
1591 if inputs.get(comp_name):
1592 comps_with_inputs.append(comp_name)
1594 # If there are no components with any inputs we fallback to checking all components in the queue.
1595 # This isn't always ideal since already executed components can also be in the queue at this point also
1596 # with blocked priority.
1597 if not comps_with_inputs:
1598 comps_with_inputs = comps_in_queue
1600 # 3. Only keep components with the lowest number of visits. Mostly needed to handle the fallback case if no
1601 # components with inputs are found.
1602 ordered_comps_with_inputs = sorted(comps_with_inputs, key=lambda x: component_visits[x])
1603 lowest_component_visit = component_visits[ordered_comps_with_inputs[0]]
1604 possible_blocking_comps = [
1605 comp for comp in ordered_comps_with_inputs if component_visits[comp] == lowest_component_visit
1606 ]
1608 # If there is only one component with the lowest visits, return it as the most likely blocking component.
1609 if len(possible_blocking_comps) == 1:
1610 blocking_comp_types = [self.graph.nodes[possible_blocking_comps[0]]["instance"].__class__.__name__]
1611 self._log_warning_for_blocking_components(
1612 blocking_comp_names=possible_blocking_comps, blocking_comp_types=blocking_comp_types
1613 )
1614 return possible_blocking_comps, blocking_comp_types
1616 # 4. Then for all components with the same lowest component visits we sort topologically before returning.
1617 topological_sort = self._topological_sort()
1618 possible_blocking_comps = sorted(
1619 possible_blocking_comps, key=lambda comp_name: (topological_sort[comp_name], comp_name.lower())
1620 )
1621 possible_blocking_comp_types = [
1622 self.graph.nodes[comp_name]["instance"].__class__.__name__ for comp_name in possible_blocking_comps
1623 ]
1625 self._log_warning_for_blocking_components(
1626 blocking_comp_names=possible_blocking_comps, blocking_comp_types=possible_blocking_comp_types
1627 )
1628 return possible_blocking_comps, possible_blocking_comp_types
1630 def _log_warning_for_blocking_components(
1631 self, blocking_comp_names: list[str], blocking_comp_types: list[dict]
1632 ) -> None:
1633 """
1634 Logs a warning about the components that are most likely blocking the pipeline execution.
1636 :param blocking_comp_names: The list of component names that are most likely blocking the pipeline execution.
1637 :param blocking_comp_types: The list of component types that are most likely blocking the pipeline execution.
1638 """
1639 comp_details = "\n".join(
1640 f" - '{name}' ({comp_type})"
1641 for name, comp_type in zip(blocking_comp_names, blocking_comp_types, strict=True)
1642 )
1643 logger.warning(
1644 "Cannot run pipeline - the pipeline appears to be blocked.\n"
1645 "The following components could not be run and may be waiting on inputs that will "
1646 "never arrive:\n" + comp_details + "\n"
1647 "Note that some of these components may be intentionally inactive due to conditional "
1648 "branching. If this is unexpected, check the connections to these components and "
1649 "ensure all required inputs are provided.",
1650 component_names=blocking_comp_names,
1651 component_types=blocking_comp_types,
1652 )
1654 def _write_component_outputs(
1655 self,
1656 *,
1657 component_name: str,
1658 component_outputs: Mapping[str, Any],
1659 inputs: InputsType,
1660 receivers: Sequence[tuple[str, OutputSocket, InputSocket, ConversionStrategyType]],
1661 include_outputs_from: set[str],
1662 ) -> Mapping[str, Any]:
1663 """
1664 Distributes the outputs of a component to the input sockets that it is connected to.
1666 :param component_name: The name of the component.
1667 :param component_outputs: The outputs of the component.
1668 :param inputs: The current global input state.
1669 :param receivers: A sequence of tuples containing:
1670 - receiver component name,
1671 - output socket of the sender,
1672 - input socket of the receiver,
1673 - ConversionStrategy to be used to convert the value if required, otherwise None.
1674 :param include_outputs_from: Set of component names that should always return an output from the pipeline.
1675 """
1676 for receiver_name, sender_socket, receiver_socket, conversion_strategy in receivers:
1677 # We either get the value that was produced by the actor or we use a _NoOutputProduced marker to indicate
1678 # that the sender did not produce an output for this socket.
1679 # This allows us to track if a predecessor already ran but did not produce an output.
1680 value = component_outputs.get(sender_socket.name, _NoOutputProduced())
1682 if not isinstance(value, _NoOutputProduced) and conversion_strategy:
1683 try:
1684 value = _convert_value(value=value, conversion_strategy=conversion_strategy)
1685 except Exception as e:
1686 sender_node = self.graph.nodes.get(component_name)
1687 sender_instance = sender_node.get("instance") if sender_node else None
1688 sender_type_name = type(sender_instance).__name__ if sender_instance else "unknown"
1690 receiver_node = self.graph.nodes.get(receiver_name)
1691 receiver_instance = receiver_node.get("instance") if receiver_node else None
1692 receiver_type_name = type(receiver_instance).__name__ if receiver_instance else "unknown"
1694 msg = (
1695 f"Failed to perform conversion between components:\n"
1696 f"Sender component: '{component_name}' (type: '{sender_type_name}')\n"
1697 f"Sender socket: '{sender_socket.name}'\n"
1698 f"Receiver component: '{receiver_name}' (type: '{receiver_type_name}')\n"
1699 f"Receiver socket: '{receiver_socket.name}'\n"
1700 f"Error: {e}"
1701 )
1702 raise PipelineRuntimeError(component_name=None, component_type=None, message=msg) from e
1704 if receiver_name not in inputs:
1705 inputs[receiver_name] = {}
1707 if receiver_socket.is_lazy_variadic:
1708 # If the receiver socket is lazy variadic, we append the new input.
1709 # Lazy variadic sockets can collect multiple inputs.
1710 _write_to_lazy_variadic_socket(
1711 inputs=inputs,
1712 receiver_name=receiver_name,
1713 receiver_socket_name=receiver_socket.name,
1714 component_name=component_name,
1715 value=value,
1716 )
1717 else:
1718 # If the receiver socket is not lazy variadic, it is greedy variadic or non-variadic.
1719 # We overwrite with the new input if it's not a _NoOutputProduced marker, or if the current value
1720 # is None.
1721 _write_to_standard_socket(
1722 inputs=inputs,
1723 receiver_name=receiver_name,
1724 receiver_socket_name=receiver_socket.name,
1725 component_name=component_name,
1726 value=value,
1727 )
1729 # If we want to include all outputs from this actor in the final outputs, we don't need to prune any consumed
1730 # outputs
1731 if component_name in include_outputs_from:
1732 return component_outputs
1734 # We prune outputs that were consumed by any receiving sockets.
1735 # All remaining outputs will be added to the final outputs of the pipeline.
1736 consumed_outputs = {sender_socket.name for _, sender_socket, __, ___ in receivers}
1737 return {key: value for key, value in component_outputs.items() if key not in consumed_outputs}
1739 @staticmethod
1740 def _is_queue_stale(priority_queue: FIFOPriorityQueue) -> bool:
1741 """
1742 Checks if the priority queue needs to be recomputed because the priorities might have changed.
1744 The queue is considered stale if it is empty or if the highest priority component is not READY or HIGHEST.
1746 For example, if the next component in a queue has the priority READY then the equality becomes
1747 ComponentPriority.READY > ComponentPriority.READY which is false.
1748 However, if the next component has priority DEFER (or BLOCKED) then the equality becomes
1749 ComponentPriority.DEFER > ComponentPriority.READY which is true, indicating that the queue is stale.
1751 :param priority_queue: Priority queue of component names.
1752 """
1753 return len(priority_queue) == 0 or priority_queue.peek()[0] > ComponentPriority.READY
1755 @staticmethod
1756 def validate_pipeline(priority_queue: FIFOPriorityQueue) -> None:
1757 """
1758 Validate the pipeline to check if it is blocked or has no valid entry point.
1760 :param priority_queue: Priority queue of component names.
1761 :raises PipelineRuntimeError:
1762 If the pipeline is blocked or has no valid entry point.
1763 """
1764 if len(priority_queue) == 0:
1765 return
1767 candidate = priority_queue.peek()
1768 if candidate is not None and candidate[0] == ComponentPriority.BLOCKED:
1769 raise PipelineComponentsBlockedError()
1771 def _find_super_components(self) -> list[tuple[str, Component]]:
1772 """
1773 Find all SuperComponents in the pipeline.
1775 :returns:
1776 List of tuples containing (component_name, component_instance) representing a SuperComponent.
1777 """
1779 super_components = []
1780 for comp_name, comp in self.walk():
1781 # a SuperComponent has a "pipeline" attribute which itself a Pipeline instance
1782 # we don't test against SuperComponent because doing so always lead to circular imports
1783 if hasattr(comp, "pipeline") and isinstance(comp.pipeline, self.__class__):
1784 super_components.append((comp_name, comp))
1785 return super_components
1787 def _merge_super_component_pipelines(self) -> tuple[networkx.MultiDiGraph, dict[str, str]]:
1788 """
1789 Merge the internal pipelines of SuperComponents into the main pipeline graph structure.
1791 This creates a new networkx.MultiDiGraph containing all the components from both the main pipeline
1792 and all the internal SuperComponents' pipelines. The SuperComponents are removed and their internal
1793 components are connected to corresponding input and output sockets of the main pipeline.
1795 :returns:
1796 A tuple containing:
1797 - A networkx.MultiDiGraph with the expanded structure of the main pipeline and all its SuperComponents
1798 - A dictionary mapping component names to their SuperComponent name
1799 """
1800 merged_graph = self.graph.copy()
1801 super_component_mapping: dict[str, str] = {}
1803 for super_name, super_component in self._find_super_components():
1804 internal_pipeline = super_component.pipeline # type: ignore
1805 internal_graph = internal_pipeline.graph.copy()
1807 # Mark all components in the internal pipeline as being part of a SuperComponent
1808 for node in internal_graph.nodes():
1809 super_component_mapping[node] = super_name
1811 # edges connected to the super component
1812 incoming_edges = list(merged_graph.in_edges(super_name, data=True))
1813 outgoing_edges = list(merged_graph.out_edges(super_name, data=True))
1815 # merge the SuperComponent graph into the main graph and remove the super component node
1816 # since its components are now part of the main graph
1817 merged_graph = networkx.compose(merged_graph, internal_graph)
1818 merged_graph.remove_node(super_name)
1820 # get the entry and exit points of the SuperComponent internal pipeline
1821 entry_points = [n for n in internal_graph.nodes() if internal_graph.in_degree(n) == 0]
1822 exit_points = [n for n in internal_graph.nodes() if internal_graph.out_degree(n) == 0]
1824 # connect the incoming edges to entry points
1825 for sender, _, edge_data in incoming_edges:
1826 sender_socket = edge_data["from_socket"]
1827 for entry_point in entry_points:
1828 # find a matching input socket in the entry point
1829 entry_point_sockets = internal_graph.nodes[entry_point]["input_sockets"]
1830 for socket_name, socket in entry_point_sockets.items():
1831 if _types_are_compatible(sender_socket.type, socket.type, self._connection_type_validation)[0]:
1832 merged_graph.add_edge(
1833 sender,
1834 entry_point,
1835 key=f"{sender_socket.name}/{socket_name}",
1836 conn_type=_type_name(sender_socket.type),
1837 from_socket=sender_socket,
1838 to_socket=socket,
1839 mandatory=socket.is_mandatory,
1840 )
1842 # connect outgoing edges from exit points
1843 for _, receiver, edge_data in outgoing_edges:
1844 receiver_socket = edge_data["to_socket"]
1845 for exit_point in exit_points:
1846 # find a matching output socket in the exit point
1847 exit_point_sockets = internal_graph.nodes[exit_point]["output_sockets"]
1848 for socket_name, socket in exit_point_sockets.items():
1849 if _types_are_compatible(socket.type, receiver_socket.type, self._connection_type_validation)[
1850 0
1851 ]:
1852 merged_graph.add_edge(
1853 exit_point,
1854 receiver,
1855 key=f"{socket_name}/{receiver_socket.name}",
1856 conn_type=_type_name(socket.type),
1857 from_socket=socket,
1858 to_socket=receiver_socket,
1859 mandatory=receiver_socket.is_mandatory,
1860 )
1862 return merged_graph, super_component_mapping
1864 def _is_pipeline_possibly_blocked(self, current_pipeline_outputs: dict[str, Any]) -> bool:
1865 """
1866 Heuristically determines whether the pipeline is possibly blocked based on its current outputs.
1868 This method checks if the pipeline has produced any of the expected outputs.
1869 - If no outputs are expected (i.e., `self.outputs()` returns an empty list), the method assumes the pipeline
1870 is not blocked.
1871 - If at least one expected output is present in `current_pipeline_outputs`, the pipeline is also assumed to not
1872 be blocked.
1873 - If none of the expected outputs are present, the pipeline is considered to be possibly blocked.
1875 Note: This check is not definitive—it is intended as a best-effort guess to detect a stalled or misconfigured
1876 pipeline when there are no more runnable components.
1878 :param current_pipeline_outputs: A dictionary of outputs currently produced by the pipeline.
1879 :returns:
1880 bool: True if the pipeline is possibly blocked (i.e., expected outputs are missing), False otherwise.
1881 """
1882 expected_outputs = self.outputs()
1883 return bool(expected_outputs) and not any(k in current_pipeline_outputs for k in expected_outputs)
1886def _connections_status(
1887 sender_node: str, receiver_node: str, sender_sockets: list[OutputSocket], receiver_sockets: list[InputSocket]
1888) -> str:
1889 """
1890 Lists the status of the sockets, for error messages.
1891 """
1892 sender_sockets_entries = []
1893 for sender_socket in sender_sockets:
1894 sender_sockets_entries.append(f" - {sender_socket.name}: {_type_name(sender_socket.type)}")
1895 sender_sockets_list = "\n".join(sender_sockets_entries)
1897 receiver_sockets_entries = []
1898 for receiver_socket in receiver_sockets:
1899 if receiver_socket.senders:
1900 sender_status = f"sent by {','.join(receiver_socket.senders)}"
1901 else:
1902 sender_status = "available"
1903 receiver_sockets_entries.append(
1904 f" - {receiver_socket.name}: {_type_name(receiver_socket.type)} ({sender_status})"
1905 )
1906 receiver_sockets_list = "\n".join(receiver_sockets_entries)
1908 return f"'{sender_node}':\n{sender_sockets_list}\n'{receiver_node}':\n{receiver_sockets_list}"
1911# Utility functions
1914def _validate_component_output_keys(
1915 component_name: str, comp: dict[str, Any], component_output: Mapping[str, Any]
1916) -> None:
1917 """
1918 Validate that the output keys returned by a component match its declared output types.
1920 Logs a warning for any actually returned output key(s) that was not declared as an output socket(s).
1921 This helps catch bugs where a component returns wrong keys, which would otherwise cause downstream components to
1922 wait forever for expected data, resulting in a confusing "Pipeline Blocked" error that points to an unexpected
1923 component.
1925 :param component_name: Name of the Component as registered in the Pipeline.
1926 :param comp: The component metadata dictionary containing the component instance and its input/output socket
1927 metadata.
1928 :param component_output: The actual output dictionary returned by the component's run method.
1929 """
1930 output_sockets = comp.get("output_sockets", {})
1931 if not output_sockets:
1932 return
1934 declared_keys = set(output_sockets.keys())
1935 actual_keys = set(component_output.keys())
1937 extra_keys = actual_keys - declared_keys
1939 instance = comp["instance"]
1940 component_type = instance.__class__.__name__
1942 if extra_keys:
1943 logger.warning(
1944 "Component '{component_name}' (type: {component_type}) returned output keys {extra_keys} "
1945 "that are not declared in its output types. "
1946 "These keys will be ignored and not passed to downstream components. "
1947 "Make sure the component's output keys match its declared @component.output_types.",
1948 component_name=component_name,
1949 component_type=component_type,
1950 extra_keys=extra_keys,
1951 )
1954def _write_to_lazy_variadic_socket(
1955 inputs: InputsType, receiver_name: str, receiver_socket_name: str, component_name: str, value: Any
1956) -> None:
1957 """
1958 Write to a lazy variadic socket.
1960 Mutates inputs in place.
1962 :param inputs: The global inputs state to be mutated.
1963 :param receiver_name: The name of the component receiving the input.
1964 :param receiver_socket_name: The name of the socket receiving the input.
1965 :param component_name: The name of the component sending the input.
1966 :param value: The value to be sent to the socket.
1967 """
1968 if not inputs[receiver_name].get(receiver_socket_name):
1969 inputs[receiver_name][receiver_socket_name] = []
1971 inputs[receiver_name][receiver_socket_name].append({"sender": component_name, "value": value})
1974def _write_to_standard_socket(
1975 inputs: InputsType, receiver_name: str, receiver_socket_name: str, component_name: str, value: Any
1976) -> None:
1977 """
1978 Write to a greedy variadic or non-variadic socket.
1980 Mutates inputs in place.
1982 :param inputs: The global inputs state to be mutated.
1983 :param receiver_name: The name of the component receiving the input.
1984 :param receiver_socket_name: The name of the socket receiving the input.
1985 :param component_name: The name of the component sending the input.
1986 :param value: The value to be sent to the socket.
1987 """
1988 current_value = inputs[receiver_name].get(receiver_socket_name)
1990 # Only overwrite if there's no existing value, or we have a new value to provide
1991 if current_value is None or not isinstance(value, _NoOutputProduced):
1992 inputs[receiver_name][receiver_socket_name] = [{"sender": component_name, "value": value}]