feat: quality of life improvement for pipeline building

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}})`.

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.

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.

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"),
+              ]
+          )
+      )

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})()

test/core/test_serialization_security.py

@@ -656,6 +656,7 @@ class TestPipelineCallableClassification:
             "close",
             "close_async",
             "connect",
+            "connect_many",
             "draw",
             "dump",
             "dumps",