diff --git a/docs-website/docs/concepts/pipelines.mdx b/docs-website/docs/concepts/pipelines.mdx index 35c742fcd8..9dea9828f0 100644 --- a/docs-website/docs/concepts/pipelines.mdx +++ b/docs-website/docs/concepts/pipelines.mdx @@ -121,7 +121,8 @@ Once all your components are created and ready to be combined in a pipeline, the 2. Add components to the pipeline with `.add_components({name: component})` or add them individually with `.add_component(name, component)`. This just adds components to the pipeline without connecting them yet. It's especially useful for loops as it allows the smooth connection of the components in the next step because they all already exist in the pipeline. -3. Connect components with `.connect("producer_component.output_name", "consumer_component.input_name")`. +3. Connect several component pairs with `.connect_many([(sender, receiver)])`, or connect one pair with + `.connect("producer_component.output_name", "consumer_component.input_name")`. At this step, you explicitly connect one of the outputs of a component to one of the inputs of the next component. This is also when the pipeline validates the connection without running the components. It makes the validation fast. 4. Run the pipeline with `.run({"component_1": {"mandatory_inputs": value}})`. Finally, you run the Pipeline by specifying the first component in the pipeline and passing its mandatory inputs. Optionally, you can pass inputs to other components, for example: `.run({"component_1": {"mandatory_inputs": value}, "component_2": {"inputs": value}})`. diff --git a/docs-website/docs/concepts/pipelines/creating-pipelines.mdx b/docs-website/docs/concepts/pipelines/creating-pipelines.mdx index a59607c2ce..a55e73c45f 100644 --- a/docs-website/docs/concepts/pipelines/creating-pipelines.mdx +++ b/docs-website/docs/concepts/pipelines/creating-pipelines.mdx @@ -94,11 +94,27 @@ query_pipeline.add_component( Adding the exact same component instance again under the same name is a no-op. A name cannot refer to a different component, and a component instance cannot appear under multiple names or belong to multiple pipelines at once. +Both `add_component()` and `add_components()` return the pipeline, so you can chain further pipeline-building calls. ### 5\. Connect components Connect the components by indicating which output of a component should be connected to the input of the next component. If a component has only one input or output and the connection is obvious, you can just pass the component name without specifying the input or output. +To create several connections in one call, pass `(sender, receiver)` pairs to `connect_many()`: + +```python +query_pipeline.connect_many( + [ + ("text_embedder.embedding", "retriever"), + ("retriever", "prompt_builder.documents"), + ("prompt_builder", "llm"), + ] +) +``` + +Haystack creates the connections in the order provided. If a pair cannot be connected, subsequent pairs are not +processed, while earlier successful connections remain in the pipeline. Repeating an existing connection is a no-op. + To understand what inputs are expected to run your pipeline, use an `.inputs()` pipeline function. See a detailed examples in the [Pipeline Inputs](#pipeline-inputs) section below. Here's a more visual explanation within the code: @@ -123,7 +139,7 @@ pipeline.connect("text_embedder.embedding", "retriever.query_embedding") pipeline.connect("text_embedder.embedding", "retriever") ``` -You need to link all the components together, connecting them gradually in pairs. Here's an explicit example for the pipeline we're assembling: +You can also connect components individually. Here's an explicit example for the pipeline we're assembling: ```python # Imagine this pipeline has four components: text_embedder, retriever, prompt_builder and llm. @@ -134,6 +150,29 @@ query_pipeline.connect("retriever", "prompt_builder.documents") query_pipeline.connect("prompt_builder", "llm") ``` +Because the component-addition and connection methods return the pipeline, you can build a pipeline fluently: + +```python +query_pipeline = ( + Pipeline() + .add_components( + { + "text_embedder": text_embedder, + "retriever": retriever, + "prompt_builder": prompt_builder, + "llm": llm, + } + ) + .connect_many( + [ + ("text_embedder.embedding", "retriever"), + ("retriever", "prompt_builder.documents"), + ("prompt_builder", "llm"), + ] + ) +) +``` + ### 6\. Run the pipeline Wait for the pipeline to validate the components and connections. If everything is OK, you can now run the pipeline. `Pipeline.run()` can be called in two ways, either passing a dictionary of the component names and their inputs, or by directly passing just the inputs. When passed directly, the pipeline resolves inputs to the correct components. diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py index a7fbe4d6f6..136534d251 100644 --- a/haystack/core/pipeline/base.py +++ b/haystack/core/pipeline/base.py @@ -13,6 +13,7 @@ from typing import Any, TextIO, TypeVar, Union, get_args import networkx +from typing_extensions import Self from haystack import logging, tracing from haystack.core.component import Component, InputSocket, OutputSocket, component @@ -393,7 +394,8 @@ def load( """ return cls.loads(fp.read(), marshaller, callbacks, allowed_modules=allowed_modules, unsafe=unsafe) - def add_component(self, name: str, instance: Component) -> None: + # Self preserves the concrete subclass for fluent calls, so Pipeline().add_component(...).run() type-checks. + def add_component(self, name: str, instance: Component) -> Self: """ Add the given component to the pipeline. @@ -405,6 +407,8 @@ def add_component(self, name: str, instance: Component) -> None: The name of the component to add. :param instance: The component instance to add. + :returns: + The Pipeline instance. :raises ValueError: If a different component with the same name already exists. @@ -414,11 +418,12 @@ def add_component(self, name: str, instance: Component) -> None: If the component instance is already in this pipeline under another name or is in another pipeline. """ if not self._validate_component(name, instance): - return + return self self._add_component_to_graph(name, instance) + return self - def add_components(self, components: Mapping[str, Component]) -> None: + def add_components(self, components: dict[str, Component]) -> Self: """ Add multiple components to the pipeline. @@ -429,7 +434,9 @@ def add_components(self, components: Mapping[str, Component]) -> None: Components already present under the same name are ignored when they are the exact same instances. :param components: - A mapping of component names to component instances. + A dictionary that maps component names to component instances. + :returns: + The Pipeline instance. :raises ValueError: If a component name is invalid or already belongs to a different component in this pipeline. @@ -442,8 +449,7 @@ def add_components(self, components: Mapping[str, Component]) -> None: components_to_add: list[tuple[str, Component]] = [] component_names_by_id: dict[int, str] = {} - # Materialize the items so custom mappings cannot change between validation and insertion. - for name, instance in list(components.items()): + for name, instance in components.items(): if not self._validate_component(name, instance): continue @@ -461,6 +467,8 @@ def add_components(self, components: Mapping[str, Component]) -> None: for name, instance in components_to_add: self._add_component_to_graph(name, instance) + return self + def _validate_component(self, name: str, instance: Component) -> bool: """Validate a component before adding it, returning whether it needs to be added.""" # Component names are unique @@ -576,7 +584,7 @@ def remove_component(self, name: str) -> Component: return instance - def connect(self, sender: str, receiver: str) -> "PipelineBase": # noqa: PLR0915 PLR0912 C901 + def connect(self, sender: str, receiver: str) -> Self: # noqa: PLR0915 PLR0912 C901 """ Connects two components together. @@ -782,6 +790,30 @@ def connect(self, sender: str, receiver: str) -> "PipelineBase": # noqa: PLR091 ) return self + def connect_many(self, connections: list[tuple[str, str]]) -> Self: + """ + Connect multiple pairs of components. + + Connections are made in the order provided. If connecting a pair raises an exception, no subsequent pairs + are connected, while earlier successful connections remain in the pipeline. Repeating an existing connection + is a no-op. + + :param connections: + A list of `(sender, receiver)` pairs. Each value uses the same format accepted by + `Pipeline.connect()`. + :returns: + The Pipeline instance. + + :raises PipelineConnectError: + If a pair of components cannot be connected. + :raises ValueError: + If a sender or receiver component is not present in the pipeline. + """ + for sender, receiver in connections: + self.connect(sender, receiver) + + return self + def get_component(self, name: str) -> Component: """ Get the component with the specified name from the pipeline. diff --git a/releasenotes/notes/pipeline-connect-many-fluent-0f5322eb79a3477e.yaml b/releasenotes/notes/pipeline-connect-many-fluent-0f5322eb79a3477e.yaml new file mode 100644 index 0000000000..ad29cf981e --- /dev/null +++ b/releasenotes/notes/pipeline-connect-many-fluent-0f5322eb79a3477e.yaml @@ -0,0 +1,25 @@ +--- +features: + - | + Add ``Pipeline.connect_many()`` to connect multiple ``(sender, receiver)`` pairs in one call. + ``Pipeline.add_component()`` and ``Pipeline.add_components()`` now return the pipeline, allowing pipeline-building + methods to be chained. + + .. code:: python + + pipeline = ( + Pipeline() + .add_components( + { + "retriever": retriever, + "prompt_builder": prompt_builder, + "llm": llm, + } + ) + .connect_many( + [ + ("retriever", "prompt_builder.documents"), + ("prompt_builder", "llm"), + ] + ) + ) diff --git a/test/core/pipeline/test_pipeline_base.py b/test/core/pipeline/test_pipeline_base.py index 5649281234..52777538b1 100644 --- a/test/core/pipeline/test_pipeline_base.py +++ b/test/core/pipeline/test_pipeline_base.py @@ -147,8 +147,8 @@ def test_add_same_component_with_same_name_is_no_op(self): pipe = PipelineBase() some_component = component_class("Some")() - pipe.add_component("some", some_component) - pipe.add_component("some", some_component) + assert pipe.add_component("some", some_component) is pipe + assert pipe.add_component("some", some_component) is pipe assert list(pipe.graph.nodes) == ["some"] assert pipe.get_component("some") is some_component @@ -175,8 +175,9 @@ def test_add_components(self): first_component = component_class("First")() second_component = component_class("Second")() - pipe.add_components({"first": first_component, "second": second_component}) + result = pipe.add_components({"first": first_component, "second": second_component}) + assert result is pipe assert list(pipe.graph.nodes) == ["first", "second"] assert pipe.get_component("first") is first_component assert pipe.get_component("second") is second_component @@ -185,8 +186,8 @@ def test_add_components_is_idempotent(self): pipe = PipelineBase() components = {"first": component_class("First")(), "second": component_class("Second")()} - pipe.add_components(components) - pipe.add_components(components) + assert pipe.add_components(components) is pipe + assert pipe.add_components(components) is pipe assert list(pipe.graph.nodes) == ["first", "second"] @@ -2060,6 +2061,49 @@ def test_from_dict_without_connection_receiver(self): err.match("Missing receiver in connection: {'sender': 'some.sender'}") +class TestPipelineConnectMany: + def test_connect_many(self): + first = component_class("First", output_types={"value": int})() + middle = component_class("Middle", input_types={"value": int}, output_types={"value": int})() + last = component_class("Last", input_types={"value": int})() + pipe = PipelineBase().add_components({"first": first, "middle": middle, "last": last}) + + result = pipe.connect_many([("first", "middle"), ("middle", "last")]) + + assert result is pipe + assert list(pipe.graph.edges) == [("first", "middle", "value/value"), ("middle", "last", "value/value")] + + def test_connect_many_with_empty_list(self): + pipe = PipelineBase() + + assert pipe.connect_many([]) is pipe + assert list(pipe.graph.edges) == [] + + def test_connect_many_is_idempotent(self): + sender = component_class("Sender", output_types={"value": int})() + receiver = component_class("Receiver", input_types={"value": int})() + pipe = PipelineBase().add_components({"sender": sender, "receiver": receiver}) + connections = [("sender.value", "receiver.value")] + + assert pipe.connect_many(connections) is pipe + assert pipe.connect_many(connections) is pipe + + assert sender.__haystack_output__.value.receivers == ["receiver"] # type: ignore[attr-defined] + assert receiver.__haystack_input__.value.senders == ["sender"] # type: ignore[attr-defined] + assert list(pipe.graph.edges) == [("sender", "receiver", "value/value")] + + def test_connect_many_stops_after_first_error(self): + first = component_class("First", output_types={"value": int})() + middle = component_class("Middle", input_types={"value": int}, output_types={"value": int})() + last = component_class("Last", input_types={"value": int})() + pipe = PipelineBase().add_components({"first": first, "middle": middle, "last": last}) + + with pytest.raises(ValueError, match="Component named missing not found"): + pipe.connect_many([("first", "middle"), ("missing", "last"), ("middle", "last")]) + + assert list(pipe.graph.edges) == [("first", "middle", "value/value")] + + class TestPipelineConnect: def test_connect(self): comp1 = component_class("Comp1", output_types={"value": int})() diff --git a/test/core/test_serialization_security.py b/test/core/test_serialization_security.py index 7b5bb8ea08..b9276ee11a 100644 --- a/test/core/test_serialization_security.py +++ b/test/core/test_serialization_security.py @@ -656,6 +656,7 @@ class TestPipelineCallableClassification: "close", "close_async", "connect", + "connect_many", "draw", "dump", "dumps",