Repository: deepset-ai/haystack

Issue:
Refactor `DocumentJoiner` to use enums for join modes
`DocumentJoiner` now takes bare strings for its `join_mode` parameter. This should rewritten to use the enum pattern seen elsewhere in the codebase.


Discussion:
coveralls: ## Pull Request Test Coverage Report for [Build 9880651127](https://coveralls.io/builds/68575907)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **1** unchanged line in **1** file lost coverage.
* Overall coverage increased (+**0.001%**) to **90.009%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/joiners/document_joiner.py](https://coveralls.io/builds/68575907/source?filename=components%2Fjoiners%2Fdocument_joiner.py#L28) | 1 | 98.96% |
<!-- | **Total:** | **1** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68575907/badge)](https://coveralls.io/builds/68575907) |
| :-- | --: |
| Change from base [Build 9874914711](https://coveralls.io/builds/68565962): |  0.001% |
| Covered Lines: | 6856 |
| Relevant Lines: | 7617 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/haystack/components/caching/cache_checker.py b/haystack/components/caching/cache_checker.py
index f4630b38..2de3ee52 100644
--- a/haystack/components/caching/cache_checker.py
+++ b/haystack/components/caching/cache_checker.py
@@ -2,10 +2,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import importlib
 from typing import Any, Dict, List
 
 from haystack import DeserializationError, Document, component, default_from_dict, default_to_dict, logging
+from haystack.core.serialization import import_class_by_name
 from haystack.document_stores.types import DocumentStore
 
 logger = logging.getLogger(__name__)
@@ -77,19 +77,13 @@ class CacheChecker:
         if "type" not in init_params["document_store"]:
             raise DeserializationError("Missing 'type' in document store's serialization data")
 
+        doc_store_data = data["init_parameters"]["document_store"]
         try:
-            module_name, type_ = init_params["document_store"]["type"].rsplit(".", 1)
-            logger.debug("Trying to import module '{module_name}'", module_name=module_name)
-            module = importlib.import_module(module_name)
-        except (ImportError, DeserializationError) as e:
-            raise DeserializationError(
-                f"DocumentStore of type '{init_params['document_store']['type']}' not correctly imported"
-            ) from e
-
-        docstore_class = getattr(module, type_)
-        docstore = docstore_class.from_dict(init_params["document_store"])
-
-        data["init_parameters"]["document_store"] = docstore
+            doc_store_class = import_class_by_name(doc_store_data["type"])
+        except ImportError as e:
+            raise DeserializationError(f"Class '{doc_store_data['type']}' not correctly imported") from e
+        data["init_parameters"]["document_store"] = default_from_dict(doc_store_class, doc_store_data)
+
         return default_from_dict(cls, data)
 
     @component.output_types(hits=List[Document], misses=List)
diff --git a/haystack/components/converters/pypdf.py b/haystack/components/converters/pypdf.py
index 9f559d08..8f5aa5ce 100644
--- a/haystack/components/converters/pypdf.py
+++ b/haystack/components/converters/pypdf.py
@@ -106,14 +106,8 @@ class PyPDFToDocument:
         :returns:
             Deserialized component.
         """
-
-        if converter := data["init_parameters"].get("converter"):
-            converter_class = deserialize_type(converter["type"])
-            data["init_parameters"]["converter"] = converter_class.from_dict(data["init_parameters"]["converter"])
-        else:
-            # Ensures backwards compatibility with Pipelines dumped with < 2.3.0
-            data["init_parameters"]["converter"] = DefaultConverter()
-
+        converter_class = deserialize_type(data["init_parameters"]["converter"]["type"])
+        data["init_parameters"]["converter"] = converter_class.from_dict(data["init_parameters"]["converter"])
         return default_from_dict(cls, data)
 
     @component.output_types(documents=List[Document])
diff --git a/haystack/components/evaluators/context_relevance.py b/haystack/components/evaluators/context_relevance.py
index b3dbae58..d6fe3adf 100644
--- a/haystack/components/evaluators/context_relevance.py
+++ b/haystack/components/evaluators/context_relevance.py
@@ -93,6 +93,7 @@ class ContextRelevanceEvaluator(LLMEvaluator):
         progress_bar: bool = True,
         api: str = "openai",
         api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"),
+        api_params: Optional[Dict[str, Any]] = None,
         raise_on_failure: bool = True,
     ):
         """
@@ -121,6 +122,8 @@ class ContextRelevanceEvaluator(LLMEvaluator):
             Supported APIs: "openai".
         :param api_key:
             The API key.
+        :param api_params:
+            Parameters for an OpenAI API compatible completions call.
         :param raise_on_failure:
             Whether to raise an exception if the API call fails.
 
@@ -137,6 +140,7 @@ class ContextRelevanceEvaluator(LLMEvaluator):
         self.examples = examples or _DEFAULT_EXAMPLES
         self.api = api
         self.api_key = api_key
+        self.api_params = api_params or {}
 
         warnings.warn(
             "The output of the ContextRelevanceEvaluator will change in Haystack 2.4.0. "
@@ -152,6 +156,7 @@ class ContextRelevanceEvaluator(LLMEvaluator):
             examples=self.examples,
             api=self.api,
             api_key=self.api_key,
+            api_params=self.api_params,
             raise_on_failure=raise_on_failure,
             progress_bar=progress_bar,
         )
@@ -202,6 +207,7 @@ class ContextRelevanceEvaluator(LLMEvaluator):
             api_key=self.api_key.to_dict() if self.api_key else None,
             examples=self.examples,
             progress_bar=self.progress_bar,
+            api_params=self.api_params,
             raise_on_failure=self.raise_on_failure,
         )
 
diff --git a/haystack/components/evaluators/faithfulness.py b/haystack/components/evaluators/faithfulness.py
index 25b38673..5cdfceaf 100644
--- a/haystack/components/evaluators/faithfulness.py
+++ b/haystack/components/evaluators/faithfulness.py
@@ -88,6 +88,7 @@ class FaithfulnessEvaluator(LLMEvaluator):
         progress_bar: bool = True,
         api: str = "openai",
         api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"),
+        api_params: Optional[Dict[str, Any]] = None,
         raise_on_failure: bool = True,
     ):
         """
@@ -117,6 +118,8 @@ class FaithfulnessEvaluator(LLMEvaluator):
             Supported APIs: "openai".
         :param api_key:
             The API key.
+        :param api_params:
+            Parameters for an OpenAI API compatible completions call.
         :param raise_on_failure:
             Whether to raise an exception if the API call fails.
 
@@ -133,6 +136,7 @@ class FaithfulnessEvaluator(LLMEvaluator):
         self.examples = examples or _DEFAULT_EXAMPLES
         self.api = api
         self.api_key = api_key
+        self.api_params = api_params or {}
 
         super(FaithfulnessEvaluator, self).__init__(
             instructions=self.instructions,
@@ -141,6 +145,7 @@ class FaithfulnessEvaluator(LLMEvaluator):
             examples=self.examples,
             api=self.api,
             api_key=self.api_key,
+            api_params=self.api_params,
             raise_on_failure=raise_on_failure,
             progress_bar=progress_bar,
         )
@@ -193,6 +198,7 @@ class FaithfulnessEvaluator(LLMEvaluator):
             self,
             api=self.api,
             api_key=self.api_key.to_dict() if self.api_key else None,
+            api_params=self.api_params,
             examples=self.examples,
             progress_bar=self.progress_bar,
             raise_on_failure=self.raise_on_failure,
diff --git a/haystack/components/evaluators/llm_evaluator.py b/haystack/components/evaluators/llm_evaluator.py
index 22885f87..511bc1cd 100644
--- a/haystack/components/evaluators/llm_evaluator.py
+++ b/haystack/components/evaluators/llm_evaluator.py
@@ -58,6 +58,7 @@ class LLMEvaluator:
         raise_on_failure: bool = True,
         api: str = "openai",
         api_key: Secret = Secret.from_env_var("OPENAI_API_KEY"),
+        api_params: Optional[Dict[str, Any]] = None,
     ):
         """
         Creates an instance of LLMEvaluator.
@@ -84,6 +85,8 @@ class LLMEvaluator:
             Supported APIs: "openai".
         :param api_key:
             The API key.
+        :param api_params:
+            Parameters for an OpenAI API compatible completions call.
 
         """
         self.validate_init_parameters(inputs, outputs, examples)
@@ -94,12 +97,16 @@ class LLMEvaluator:
         self.examples = examples
         self.api = api
         self.api_key = api_key
+        self.api_params = api_params or {}
         self.progress_bar = progress_bar
 
+        default_generation_kwargs = {"response_format": {"type": "json_object"}, "seed": 42}
+        user_generation_kwargs = self.api_params.get("generation_kwargs", {})
+        merged_generation_kwargs = {**default_generation_kwargs, **user_generation_kwargs}
+        self.api_params["generation_kwargs"] = merged_generation_kwargs
+
         if api == "openai":
-            self.generator = OpenAIGenerator(
-                api_key=api_key, generation_kwargs={"response_format": {"type": "json_object"}, "seed": 42}
-            )
+            self.generator = OpenAIGenerator(api_key=api_key, **self.api_params)
         else:
             raise ValueError(f"Unsupported API: {api}")
 
@@ -282,6 +289,7 @@ class LLMEvaluator:
             examples=self.examples,
             api=self.api,
             api_key=self.api_key.to_dict(),
+            api_params=self.api_params,
             progress_bar=self.progress_bar,
         )
 
diff --git a/haystack/components/joiners/document_joiner.py b/haystack/components/joiners/document_joiner.py
index bd51613d..c3bf78cb 100644
--- a/haystack/components/joiners/document_joiner.py
+++ b/haystack/components/joiners/document_joiner.py
@@ -4,15 +4,42 @@
 
 import itertools
 from collections import defaultdict
+from enum import Enum
 from math import inf
-from typing import List, Optional
+from typing import Any, Dict, List, Optional, Union
 
-from haystack import Document, component, logging
+from haystack import Document, component, default_from_dict, default_to_dict, logging
 from haystack.core.component.types import Variadic
 
 logger = logging.getLogger(__name__)
 
 
+class JoinMode(Enum):
+    """
+    Enum for join mode.
+    """
+
+    CONCATENATE = "concatenate"
+    MERGE = "merge"
+    RECIPROCAL_RANK_FUSION = "reciprocal_rank_fusion"
+    DISTRIBUTION_BASED_RANK_FUSION = "distribution_based_rank_fusion"
+
+    def __str__(self):
+        return self.value
+
+    @staticmethod
+    def from_str(string: str) -> "JoinMode":
+        """
+        Convert a string to a JoinMode enum.
+        """
+        enum_map = {e.value: e for e in JoinMode}
+        mode = enum_map.get(string)
+        if mode is None:
+            msg = f"Unknown join mode '{string}'. Supported modes in DocumentJoiner are: {list(enum_map.keys())}"
+            raise ValueError(msg)
+        return mode
+
+
 @component
 class DocumentJoiner:
     """
@@ -45,7 +72,7 @@ class DocumentJoiner:
 
     def __init__(
         self,
-        join_mode: str = "concatenate",
+        join_mode: Union[str, JoinMode] = JoinMode.CONCATENATE,
         weights: Optional[List[float]] = None,
         top_k: Optional[int] = None,
         sort_by_score: bool = True,
@@ -68,8 +95,15 @@ class DocumentJoiner:
             If True sorts the Documents by score in descending order.
             If a Document has no score, it is handled as if its score is -infinity.
         """
-        if join_mode not in ["concatenate", "merge", "reciprocal_rank_fusion", "distribution_based_rank_fusion"]:
-            raise ValueError(f"DocumentJoiner component does not support '{join_mode}' join_mode.")
+        if isinstance(join_mode, str):
+            join_mode = JoinMode.from_str(join_mode)
+        join_mode_functions = {
+            JoinMode.CONCATENATE: self._concatenate,
+            JoinMode.MERGE: self._merge,
+            JoinMode.RECIPROCAL_RANK_FUSION: self._reciprocal_rank_fusion,
+            JoinMode.DISTRIBUTION_BASED_RANK_FUSION: self._distribution_based_rank_fusion,
+        }
+        self.join_mode_function = join_mode_functions[join_mode]
         self.join_mode = join_mode
         self.weights = [float(i) / sum(weights) for i in weights] if weights else None
         self.top_k = top_k
@@ -92,14 +126,7 @@ class DocumentJoiner:
         output_documents = []
 
         documents = list(documents)
-        if self.join_mode == "concatenate":
-            output_documents = self._concatenate(documents)
-        elif self.join_mode == "merge":
-            output_documents = self._merge(documents)
-        elif self.join_mode == "reciprocal_rank_fusion":
-            output_documents = self._reciprocal_rank_fusion(documents)
-        elif self.join_mode == "distribution_based_rank_fusion":
-            output_documents = self._distribution_based_rank_fusion(documents)
+        output_documents = self.join_mode_function(documents)
 
         if self.sort_by_score:
             output_documents = sorted(
@@ -204,3 +231,30 @@ class DocumentJoiner:
         output = self._concatenate(document_lists=document_lists)
 
         return output
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Serializes the component to a dictionary.
+
+        :returns:
+            Dictionary with serialized data.
+        """
+        return default_to_dict(
+            self,
+            join_mode=str(self.join_mode),
+            weights=self.weights,
+            top_k=self.top_k,
+            sort_by_score=self.sort_by_score,
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "DocumentJoiner":
+        """
+        Deserializes the component from a dictionary.
+
+        :param data:
+            The dictionary to deserialize from.
+        :returns:
+            The deserialized component.
+        """
+        return default_from_dict(cls, data)
diff --git a/haystack/components/retrievers/filter_retriever.py b/haystack/components/retrievers/filter_retriever.py
index b74c7b14..51e293f4 100644
--- a/haystack/components/retrievers/filter_retriever.py
+++ b/haystack/components/retrievers/filter_retriever.py
@@ -2,10 +2,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import importlib
 from typing import Any, Dict, List, Optional
 
 from haystack import DeserializationError, Document, component, default_from_dict, default_to_dict, logging
+from haystack.core.serialization import import_class_by_name
 from haystack.document_stores.types import DocumentStore
 
 logger = logging.getLogger(__name__)
@@ -81,17 +81,14 @@ class FilterRetriever:
             raise DeserializationError("Missing 'document_store' in serialization data")
         if "type" not in init_params["document_store"]:
             raise DeserializationError("Missing 'type' in document store's serialization data")
+
+        doc_store_data = data["init_parameters"]["document_store"]
         try:
-            module_name, type_ = init_params["document_store"]["type"].rsplit(".", 1)
-            logger.debug("Trying to import module '{module_name}'", module_name=module_name)
-            module = importlib.import_module(module_name)
-        except (ImportError, DeserializationError) as e:
-            raise DeserializationError(
-                f"DocumentStore of type '{init_params['document_store']['type']}' not correctly imported"
-            ) from e
-
-        docstore_class = getattr(module, type_)
-        data["init_parameters"]["document_store"] = docstore_class.from_dict(data["init_parameters"]["document_store"])
+            doc_store_class = import_class_by_name(doc_store_data["type"])
+        except ImportError as e:
+            raise DeserializationError(f"Class '{doc_store_data['type']}' not correctly imported") from e
+        data["init_parameters"]["document_store"] = default_from_dict(doc_store_class, doc_store_data)
+
         return default_from_dict(cls, data)
 
     @component.output_types(documents=List[Document])
diff --git a/haystack/components/writers/document_writer.py b/haystack/components/writers/document_writer.py
index 44b8528f..1f384bb5 100644
--- a/haystack/components/writers/document_writer.py
+++ b/haystack/components/writers/document_writer.py
@@ -2,10 +2,10 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-import importlib
 from typing import Any, Dict, List, Optional
 
 from haystack import DeserializationError, Document, component, default_from_dict, default_to_dict, logging
+from haystack.core.serialization import import_class_by_name
 from haystack.document_stores.types import DocumentStore, DuplicatePolicy
 
 logger = logging.getLogger(__name__)
@@ -79,20 +79,14 @@ class DocumentWriter:
         if "type" not in init_params["document_store"]:
             raise DeserializationError("Missing 'type' in document store's serialization data")
 
+        doc_store_data = data["init_parameters"]["document_store"]
         try:
-            module_name, type_ = init_params["document_store"]["type"].rsplit(".", 1)
-            logger.debug("Trying to import module '{module_name}'", module_name=module_name)
-            module = importlib.import_module(module_name)
-        except (ImportError, DeserializationError) as e:
-            raise DeserializationError(
-                f"DocumentStore of type '{init_params['document_store']['type']}' not correctly imported"
-            ) from e
-
-        docstore_class = getattr(module, type_)
-        docstore = docstore_class.from_dict(init_params["document_store"])
-
-        data["init_parameters"]["document_store"] = docstore
+            doc_store_class = import_class_by_name(doc_store_data["type"])
+        except ImportError as e:
+            raise DeserializationError(f"Class '{doc_store_data['type']}' not correctly imported") from e
+        data["init_parameters"]["document_store"] = default_from_dict(doc_store_class, doc_store_data)
         data["init_parameters"]["policy"] = DuplicatePolicy[data["init_parameters"]["policy"]]
+
         return default_from_dict(cls, data)
 
     @component.output_types(documents_written=int)
diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py
index ef4857d6..b1ef5fed 100644
--- a/haystack/core/pipeline/base.py
+++ b/haystack/core/pipeline/base.py
@@ -8,7 +8,7 @@ from collections import defaultdict
 from copy import copy, deepcopy
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Dict, Iterator, List, Optional, TextIO, Tuple, Type, TypeVar, Union
+from typing import Any, Dict, Iterator, List, Optional, Set, TextIO, Tuple, Type, TypeVar, Union
 
 import networkx  # type:ignore
 
@@ -765,28 +765,28 @@ class PipelineBase:
 
         return {**data}
 
-    def _init_to_run(self, pipeline_inputs: Dict[str, Any]) -> List[Tuple[str, Component]]:
-        to_run: List[Tuple[str, Component]] = []
+    def _init_run_queue(self, pipeline_inputs: Dict[str, Any]) -> List[Tuple[str, Component]]:
+        run_queue: List[Tuple[str, Component]] = []
         for node_name in self.graph.nodes:
             component = self.graph.nodes[node_name]["instance"]
 
             if len(component.__haystack_input__._sockets_dict) == 0:
                 # Component has no input, can run right away
-                to_run.append((node_name, component))
+                run_queue.append((node_name, component))
                 continue
 
             if node_name in pipeline_inputs:
                 # This component is in the input data, if it has enough inputs it can run right away
-                to_run.append((node_name, component))
+                run_queue.append((node_name, component))
                 continue
 
             for socket in component.__haystack_input__._sockets_dict.values():
                 if not socket.senders or socket.is_variadic:
                     # Component has at least one input not connected or is variadic, can run right away.
-                    to_run.append((node_name, component))
+                    run_queue.append((node_name, component))
                     break
 
-        return to_run
+        return run_queue
 
     @classmethod
     def from_template(
@@ -820,43 +820,13 @@ class PipelineBase:
         for node in self.graph.nodes:
             self.graph.nodes[node]["visits"] = 0
 
-    def _dequeue_components_that_received_no_input(
-        self,
-        component_name: str,
-        component_result: Dict[str, Any],
-        to_run: List[Tuple[str, Component]],
-        waiting_for_input: List[Tuple[str, Component]],
-    ):
-        """
-        Removes Components that didn't receive any input from the list of Components to run.
-
-        We can't run those Components if they didn't receive any input, even if it's optional.
-        This is mainly useful for Components that have conditional outputs.
-
-        :param component_name: Name of the Component that created the output
-        :param component_result: The output of the Component
-        :param to_run: Queue of Components to run
-        :param waiting_for_input: Queue of Components waiting for input
-        """
-        instance: Component = self.graph.nodes[component_name]["instance"]
-        for socket_name, socket in instance.__haystack_output__._sockets_dict.items():  # type: ignore
-            if socket_name in component_result:
-                continue
-            for receiver in socket.receivers:
-                receiver_instance: Component = self.graph.nodes[receiver]["instance"]
-                pair = (receiver, receiver_instance)
-                if pair in to_run:
-                    to_run.remove(pair)
-                if pair in waiting_for_input:
-                    waiting_for_input.remove(pair)
-
     def _distribute_output(
         self,
         component_name: str,
         component_result: Dict[str, Any],
-        inputs_by_component: Dict[str, Dict[str, Any]],
-        to_run: List[Tuple[str, Component]],
-        waiting_for_input: List[Tuple[str, Component]],
+        components_inputs: Dict[str, Dict[str, Any]],
+        run_queue: List[Tuple[str, Component]],
+        waiting_queue: List[Tuple[str, Component]],
     ) -> Dict[str, Any]:
         """
         Distributes the output of a Component to the next Components that need it.
@@ -866,11 +836,11 @@ class PipelineBase:
 
         :param component_name: Name of the Component that created the output
         :param component_result: The output of the Component
-        :paramt inputs_by_component: The current state of the inputs divided by Component name
-        :param to_run: Queue of Components to run
-        :param waiting_for_input: Queue of Components waiting for input
+        :paramt components_inputs: The current state of the inputs divided by Component name
+        :param run_queue: Queue of Components to run
+        :param waiting_queue: Queue of Components waiting for input
 
-        :return: The updated output of the Component without the keys that were distributed to other Components
+        :returns: The updated output of the Component without the keys that were distributed to other Components
         """
         # We keep track of which keys to remove from component_result at the end of the loop.
         # This is done after the output has been distributed to the next components, so that
@@ -892,8 +862,8 @@ class PipelineBase:
                 # Though it will return only one output at a time.
                 continue
 
-            if receiver_name not in inputs_by_component:
-                inputs_by_component[receiver_name] = {}
+            if receiver_name not in components_inputs:
+                components_inputs[receiver_name] = {}
 
             # We keep track of the keys that were distributed to other Components.
             # This key will be removed from component_result at the end of the loop.
@@ -906,15 +876,15 @@ class PipelineBase:
                 # instead to receive inputs from multiple senders.
                 #
                 # To keep track of all the inputs received internally we always store them in a list.
-                if receiver_socket.name not in inputs_by_component[receiver_name]:
+                if receiver_socket.name not in components_inputs[receiver_name]:
                     # Create the list if it doesn't exist
-                    inputs_by_component[receiver_name][receiver_socket.name] = []
+                    components_inputs[receiver_name][receiver_socket.name] = []
                 else:
                     # Check if the value is actually a list
-                    assert isinstance(inputs_by_component[receiver_name][receiver_socket.name], list)
-                inputs_by_component[receiver_name][receiver_socket.name].append(value)
+                    assert isinstance(components_inputs[receiver_name][receiver_socket.name], list)
+                components_inputs[receiver_name][receiver_socket.name].append(value)
             else:
-                inputs_by_component[receiver_name][receiver_socket.name] = value
+                components_inputs[receiver_name][receiver_socket.name] = value
 
             receiver = self.graph.nodes[receiver_name]["instance"]
             pair = (receiver_name, receiver)
@@ -923,111 +893,167 @@ class PipelineBase:
             if receiver_socket.is_variadic and is_greedy:
                 # If the receiver is greedy, we can run it as soon as possible.
                 # First we remove it from the status lists it's in if it's there or we risk running it multiple times.
-                if pair in to_run:
-                    to_run.remove(pair)
-                if pair in waiting_for_input:
-                    waiting_for_input.remove(pair)
-                to_run.append(pair)
+                if pair in run_queue:
+                    run_queue.remove(pair)
+                if pair in waiting_queue:
+                    waiting_queue.remove(pair)
+                run_queue.append(pair)
 
-            if pair not in waiting_for_input and pair not in to_run:
+            if pair not in waiting_queue and pair not in run_queue:
                 # Queue up the Component that received this input to run, only if it's not already waiting
                 # for input or already ready to run.
-                to_run.append(pair)
+                run_queue.append(pair)
 
         # Returns the output without the keys that were distributed to other Components
         return {k: v for k, v in component_result.items() if k not in to_remove_from_component_result}
 
-    def _enqueue_next_runnable_component(
-        self,
-        inputs_by_component: Dict[str, Dict[str, Any]],
-        to_run: List[Tuple[str, Component]],
-        waiting_for_input: List[Tuple[str, Component]],
-    ):
-        """
-        Finds the next Component that can be run and adds it to the queue of Components to run.
-
-        :param inputs_by_component: The current state of the inputs divided by Component name
-        :param to_run: Queue of Components to run
-        :param waiting_for_input: Queue of Components waiting for input
+    def _find_next_runnable_component(
+        self, components_inputs: Dict[str, Dict[str, Any]], waiting_queue: List[Tuple[str, Component]]
+    ) -> Tuple[str, Component]:
         """
-        for name, comp in waiting_for_input:
-            if name not in inputs_by_component:
-                inputs_by_component[name] = {}
-
-        # Small utility function to check if a Component has a Variadic input that is not greedy.
-        def is_lazy_variadic(c: Component) -> bool:
-            is_variadic = any(
-                socket.is_variadic
-                for socket in c.__haystack_input__._sockets_dict.values()  # type: ignore
-            )
-            if not is_variadic:
-                return False
-            return not getattr(c, "__haystack_is_greedy__", False)
-
-        # Small utility function to check if a Component has all inputs with defaults.
-        def has_all_inputs_with_defaults(c: Component) -> bool:
-            return all(
-                not socket.is_mandatory
-                for socket in c.__haystack_input__._sockets_dict.values()  # type: ignore
-            )
+        Finds the next Component that can be run and returns it.
 
-        # Updates the inputs with the default values for the inputs that are missing
-        def add_missing_input_defaults(name: str, comp: Component, inputs_by_component: Dict[str, Dict[str, Any]]):
-            for input_socket in comp.__haystack_input__._sockets_dict.values():  # type: ignore
-                if input_socket.name not in inputs_by_component[name]:
-                    inputs_by_component[name][input_socket.name] = input_socket.default_value
+        :param components_inputs: The current state of the inputs divided by Component name
+        :param waiting_queue: Queue of Components waiting for input
 
+        :returns: The name and the instance of the next Component that can be run
+        """
         all_lazy_variadic = True
         all_with_default_inputs = True
 
-        filtered_waiting_for_input = []
+        filtered_waiting_queue = []
 
-        for name, comp in waiting_for_input:
-            if not is_lazy_variadic(comp):
+        for name, comp in waiting_queue:
+            if not _is_lazy_variadic(comp):
                 # Components with variadic inputs that are not greedy must be removed only if there's nothing else to
                 # run at this stage.
                 # We need to wait as long as possible to run them, so we can collect as most inputs as we can.
                 all_lazy_variadic = False
 
-            if not has_all_inputs_with_defaults(comp):
+            if not _has_all_inputs_with_defaults(comp):
                 # Components that have defaults for all their inputs must be treated the same identical way as we treat
                 # lazy variadic components. If there are only components with defaults we can run them.
                 # If we don't do this the order of execution of the Pipeline's Components will be affected cause we
-                # enqueue the Components in `to_run` at the start using the order they are added in the Pipeline.
+                # enqueue the Components in `run_queue` at the start using the order they are added in the Pipeline.
                 # If a Component A with defaults is added before a Component B that has no defaults, but in the Pipeline
-                # logic A must be executed after B it could run instead before if we don't do this check.
+                # logic A must be executed after B. However, B could run before A if we don't do this check.
                 all_with_default_inputs = False
 
-            if not is_lazy_variadic(comp) and not has_all_inputs_with_defaults(comp):
+            if not _is_lazy_variadic(comp) and not _has_all_inputs_with_defaults(comp):
                 # Keep track of the Components that are not lazy variadic and don't have all inputs with defaults.
                 # We'll handle these later if necessary.
-                filtered_waiting_for_input.append((name, comp))
+                filtered_waiting_queue.append((name, comp))
 
         # If all Components are lazy variadic or all Components have all inputs with defaults we can get one to run
         if all_lazy_variadic or all_with_default_inputs:
-            pair = waiting_for_input.pop(0)
-            to_run.append(pair)
-            # Add missing input defaults if needed, this is a no-op for Components with Variadic inputs
-            add_missing_input_defaults(name, comp, inputs_by_component)
-            return
+            return waiting_queue[0]
 
-        for name, comp in filtered_waiting_for_input:
+        for name, comp in filtered_waiting_queue:
             # Find the first component that has all the inputs it needs to run
             has_enough_inputs = True
             for input_socket in comp.__haystack_input__._sockets_dict.values():  # type: ignore
-                if input_socket.name not in inputs_by_component[name]:
-                    if input_socket.is_mandatory:
-                        has_enough_inputs = False
-                        break
-
-                    if input_socket.name not in inputs_by_component[name]:
-                        inputs_by_component[name][input_socket.name] = input_socket.default_value
+                if input_socket.name not in components_inputs.get(name, {}) and input_socket.is_mandatory:
+                    has_enough_inputs = False
+                    break
 
             if has_enough_inputs:
+                return name, comp
+
+        # If we reach this point it means that we found no Component that has enough inputs to run.
+        # Ideally we should never reach this point, though we can't raise an exception either as
+        # existing use cases rely on this behavior.
+        # So we return the last Component, that could be the last from waiting_queue or filtered_waiting_queue.
+        return name, comp
+
+    def _find_next_runnable_lazy_variadic_or_default_component(
+        self, waiting_queue: List[Tuple[str, Component]]
+    ) -> Tuple[str, Component]:
+        """
+        Finds the next Component that can be run and has a lazy variadic input or all inputs with default values.
+
+        :param waiting_queue: Queue of Components waiting for input
+
+        :returns: The name and the instance of the next Component that can be run
+        """
+        for name, comp in waiting_queue:
+            is_lazy_variadic = _is_lazy_variadic(comp)
+            has_only_defaults = _has_all_inputs_with_defaults(comp)
+            if is_lazy_variadic or has_only_defaults:
+                return name, comp
+
+        # If we reach this point it means that we found no Component that has a lazy variadic input or all inputs with
+        # default values to run.
+        # Similar to `_find_next_runnable_component` we might not find the Component we want, so we optimistically
+        # return the last Component in the list.
+        # We're probably stuck in a loop in this case, but we can't raise an exception as existing use cases might
+        # rely on this behaviour.
+        # The loop detection will be handled later on.
+        return name, comp
+
+    def _find_components_that_received_no_input(
+        self, component_name: str, component_result: Dict[str, Any]
+    ) -> Set[Tuple[str, Component]]:
+        """
+        Find all the Components that are connected to component_name and didn't receive any input from it.
+
+        :param component_name: Name of the Component that created the output
+        :param component_result: Output of the Component
+        :return: A set of Components that didn't receive any input from component_name
+        """
+        components = set()
+        instance: Component = self.graph.nodes[component_name]["instance"]
+        for socket_name, socket in instance.__haystack_output__._sockets_dict.items():  # type: ignore
+            if socket_name in component_result:
+                continue
+            for receiver in socket.receivers:
+                receiver_instance: Component = self.graph.nodes[receiver]["instance"]
+                components.add((receiver, receiver_instance))
+        return components
+
+    def _is_stuck_in_a_loop(self, waiting_queue: List[Tuple[str, Component]]) -> bool:
+        """
+        Checks if the Pipeline is stuck in a loop.
+
+        :param waiting_queue: Queue of Components waiting for input
+
+        :returns: True if the Pipeline is stuck in a loop, False otherwise
+        """
+        # Are we actually stuck or there's a lazy variadic or a component with has only default inputs
+        # waiting for input?
+        # This is our last resort, if there's no lazy variadic or component with only default inputs
+        # waiting for input we're stuck for real and we can't make any progress.
+        component_found = False
+        for _, comp in waiting_queue:
+            if _is_lazy_variadic(comp) or _has_all_inputs_with_defaults(comp):
+                component_found = True
                 break
 
-        waiting_for_input.remove((name, comp))
-        to_run.append((name, comp))
+        if not component_found:
+            # We're stuck in a loop for real, we can't make any progress.
+            # BAIL!
+            return True
+
+        # If we have a single component with no variadic input or only default inputs waiting for input
+        # it means it has been waiting for input for at least 2 iterations.
+        # This will never run.
+        # BAIL!
+        return len(waiting_queue) == 1
+
+    def _component_has_enough_inputs_to_run(self, name: str, inputs: Dict[str, Dict[str, Any]]) -> bool:
+        """
+        Returns True if the Component has all the inputs it needs to run.
+
+        :param name: Name of the Component as defined in the Pipeline.
+        :param inputs: The current state of the inputs divided by Component name.
+
+        :return: Whether the Component can run or not.
+        """
+        instance: Component = self.graph.nodes[name]["instance"]
+        if name not in inputs:
+            return False
+        expected_inputs = instance.__haystack_input__._sockets_dict.keys()  # type: ignore
+        current_inputs = inputs[name].keys()
+        return expected_inputs == current_inputs
 
 
 def _connections_status(
@@ -1053,3 +1079,107 @@ def _connections_status(
     receiver_sockets_list = "\n".join(receiver_sockets_entries)
 
     return f"'{sender_node}':\n{sender_sockets_list}\n'{receiver_node}':\n{receiver_sockets_list}"
+
+
+def _is_lazy_variadic(c: Component) -> bool:
+    """
+    Small utility function to check if a Component has a Variadic input that is not greedy
+    """
+    is_variadic = any(
+        socket.is_variadic
+        for socket in c.__haystack_input__._sockets_dict.values()  # type: ignore
+    )
+    if not is_variadic:
+        return False
+    return not getattr(c, "__haystack_is_greedy__", False)
+
+
+def _has_all_inputs_with_defaults(c: Component) -> bool:
+    """
+    Small utility function to check if a Component has all inputs with defaults.
+    """
+    return all(
+        not socket.is_mandatory
+        for socket in c.__haystack_input__._sockets_dict.values()  # type: ignore
+    )
+
+
+def _add_missing_input_defaults(name: str, comp: Component, components_inputs: Dict[str, Dict[str, Any]]):
+    """
+    Updates the inputs with the default values for the inputs that are missing
+
+    :param name: Name of the Component
+    :param comp: Instance of the Component
+    :param components_inputs: The current state of the inputs divided by Component name
+    """
+    if name not in components_inputs:
+        components_inputs[name] = {}
+
+    for input_socket in comp.__haystack_input__._sockets_dict.values():  # type: ignore
+        if input_socket.is_mandatory:
+            continue
+
+        if input_socket.name not in components_inputs[name]:
+            components_inputs[name][input_socket.name] = input_socket.default_value
+
+
+def _enqueue_component(
+    component_pair: Tuple[str, Component],
+    run_queue: List[Tuple[str, Component]],
+    waiting_queue: List[Tuple[str, Component]],
+):
+    """
+    Append a Component in the queue of Components to run if not already in it.
+
+    Remove it from the waiting list if it's there.
+
+    :param component_pair: Tuple of Component name and instance
+    :param run_queue: Queue of Components to run
+    :param waiting_queue: Queue of Components waiting for input
+    """
+    if component_pair in waiting_queue:
+        waiting_queue.remove(component_pair)
+
+    if component_pair not in run_queue:
+        run_queue.append(component_pair)
+
+
+def _dequeue_component(
+    component_pair: Tuple[str, Component],
+    run_queue: List[Tuple[str, Component]],
+    waiting_queue: List[Tuple[str, Component]],
+):
+    """
+    Removes a Component both from the queue of Components to run and the waiting list.
+
+    :param component_pair: Tuple of Component name and instance
+    :param run_queue: Queue of Components to run
+    :param waiting_queue: Queue of Components waiting for input
+    """
+    if component_pair in waiting_queue:
+        waiting_queue.remove(component_pair)
+
+    if component_pair in run_queue:
+        run_queue.remove(component_pair)
+
+
+def _enqueue_waiting_component(component_pair: Tuple[str, Component], waiting_queue: List[Tuple[str, Component]]):
+    """
+    Append a Component in the queue of Components that are waiting for inputs if not already in it.
+
+    :param component_pair: Tuple of Component name and instance
+    :param waiting_queue: Queue of Components waiting for input
+    """
+    if component_pair not in waiting_queue:
+        waiting_queue.append(component_pair)
+
+
+def _dequeue_waiting_component(component_pair: Tuple[str, Component], waiting_queue: List[Tuple[str, Component]]):
+    """
+    Removes a Component from the queue of Components that are waiting for inputs.
+
+    :param component_pair: Tuple of Component name and instance
+    :param waiting_queue: Queue of Components waiting for input
+    """
+    if component_pair in waiting_queue:
+        waiting_queue.remove(component_pair)
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index 360f0f6d..0ee88c9b 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -4,13 +4,20 @@
 
 from copy import deepcopy
 from typing import Any, Dict, List, Mapping, Optional, Set, Tuple
+from warnings import warn
 
 from haystack import logging, tracing
 from haystack.core.component import Component
 from haystack.core.errors import PipelineMaxLoops, PipelineRuntimeError
+from haystack.core.pipeline.base import (
+    _dequeue_component,
+    _dequeue_waiting_component,
+    _enqueue_component,
+    _enqueue_waiting_component,
+)
 from haystack.telemetry import pipeline_running
 
-from .base import PipelineBase
+from .base import PipelineBase, _add_missing_input_defaults, _is_lazy_variadic
 
 logger = logging.getLogger(__name__)
 
@@ -22,22 +29,6 @@ class Pipeline(PipelineBase):
     Orchestrates component execution according to the execution graph, one after the other.
     """
 
-    def _component_has_enough_inputs_to_run(self, name: str, inputs: Dict[str, Dict[str, Any]]) -> bool:
-        """
-        Returns True if the Component has all the inputs it needs to run.
-
-        :param name: Name of the Component as defined in the Pipeline.
-        :param inputs: The current state of the inputs divided by Component name.
-
-        :return: Whether the Component can run or not.
-        """
-        instance: Component = self.graph.nodes[name]["instance"]
-        if name not in inputs:
-            return False
-        expected_inputs = instance.__haystack_input__._sockets_dict.keys()  # type: ignore
-        current_inputs = inputs[name].keys()
-        return expected_inputs == current_inputs
-
     def _run_component(self, name: str, inputs: Dict[str, Any]) -> Dict[str, Any]:
         """
         Runs a Component with the given inputs.
@@ -93,9 +84,7 @@ class Pipeline(PipelineBase):
 
             return res
 
-    # TODO: We're ignoring these linting rules for the time being, after we properly optimize this function we'll
-    # remove the noqa
-    def run(  # noqa: C901, PLR0912, PLR0915 pylint: disable=too-many-branches,too-many-locals
+    def run(  # noqa: PLR0915
         self, data: Dict[str, Any], debug: bool = False, include_outputs_from: Optional[Set[str]] = None
     ) -> Dict[str, Any]:
         """
@@ -182,26 +171,30 @@ class Pipeline(PipelineBase):
         self._validate_input(data)
 
         # Initialize the inputs state
-        last_inputs: Dict[str, Dict[str, Any]] = self._init_inputs_state(data)
+        components_inputs: Dict[str, Dict[str, Any]] = self._init_inputs_state(data)
 
         # Take all components that:
         # - have no inputs
         # - receive input from the user
         # - have at least one input not connected
         # - have at least one input that is variadic
-        to_run: List[Tuple[str, Component]] = self._init_to_run(data)
+        run_queue: List[Tuple[str, Component]] = self._init_run_queue(data)
 
         # These variables are used to detect when we're stuck in a loop.
         # Stuck loops can happen when one or more components are waiting for input but
         # no other component is going to run.
         # This can happen when a whole branch of the graph is skipped for example.
-        # When we find that two consecutive iterations of the loop where the waiting_for_input list is the same,
+        # When we find that two consecutive iterations of the loop where the waiting_queue is the same,
         # we know we're stuck in a loop and we can't make any progress.
-        before_last_waiting_for_input: Optional[Set[str]] = None
-        last_waiting_for_input: Optional[Set[str]] = None
+        #
+        # They track the previous two states of the waiting_queue. So if waiting_queue would n,
+        # before_last_waiting_queue would be n-2 and last_waiting_queue would be n-1.
+        # When we run a component, we reset both.
+        before_last_waiting_queue: Optional[Set[str]] = None
+        last_waiting_queue: Optional[Set[str]] = None
 
         # The waiting_for_input list is used to keep track of components that are waiting for input.
-        waiting_for_input: List[Tuple[str, Component]] = []
+        waiting_queue: List[Tuple[str, Component]] = []
 
         include_outputs_from = set() if include_outputs_from is None else include_outputs_from
 
@@ -221,32 +214,21 @@ class Pipeline(PipelineBase):
             # Cache for extra outputs, if enabled.
             extra_outputs: Dict[Any, Any] = {}
 
-            while len(to_run) > 0:
-                name, comp = to_run.pop(0)
-
-                if any(socket.is_variadic for socket in comp.__haystack_input__._sockets_dict.values()) and not getattr(  # type: ignore
-                    comp, "is_greedy", False
-                ):
-                    there_are_non_variadics = False
-                    for _, other_comp in to_run:
-                        if not any(
-                            socket.is_variadic
-                            for socket in other_comp.__haystack_input__._sockets_dict.values()  # type: ignore
-                        ):
-                            there_are_non_variadics = True
-                            break
+            while len(run_queue) > 0:
+                name, comp = run_queue.pop(0)
 
-                    if there_are_non_variadics:
-                        if (name, comp) not in waiting_for_input:
-                            waiting_for_input.append((name, comp))
-                        continue
+                if _is_lazy_variadic(comp) and not all(_is_lazy_variadic(comp) for _, comp in run_queue):
+                    # We run Components with lazy variadic inputs only if there only Components with
+                    # lazy variadic inputs left to run
+                    _enqueue_waiting_component((name, comp), waiting_queue)
+                    continue
 
-                if self._component_has_enough_inputs_to_run(name, last_inputs):
+                if self._component_has_enough_inputs_to_run(name, components_inputs):
                     if self.graph.nodes[name]["visits"] > self.max_loops_allowed:
                         msg = f"Maximum loops count ({self.max_loops_allowed}) exceeded for component '{name}'"
                         raise PipelineMaxLoops(msg)
 
-                    res: Dict[str, Any] = self._run_component(name, last_inputs[name])
+                    res: Dict[str, Any] = self._run_component(name, components_inputs[name])
 
                     if name in include_outputs_from:
                         # Deepcopy the outputs to prevent downstream nodes from modifying them
@@ -254,80 +236,52 @@ class Pipeline(PipelineBase):
                         extra_outputs[name] = deepcopy(res)
 
                     # Reset the waiting for input previous states, we managed to run a component
-                    before_last_waiting_for_input = None
-                    last_waiting_for_input = None
+                    before_last_waiting_queue = None
+                    last_waiting_queue = None
 
-                    if (name, comp) in waiting_for_input:
-                        # We manage to run this component that was in the waiting list, we can remove it.
-                        # This happens when a component was put in the waiting list but we reached it from another edge.
-                        waiting_for_input.remove((name, comp))
+                    # We manage to run this component that was in the waiting list, we can remove it.
+                    # This happens when a component was put in the waiting list but we reached it from another edge.
+                    _dequeue_waiting_component((name, comp), waiting_queue)
 
-                    self._dequeue_components_that_received_no_input(name, res, to_run, waiting_for_input)
-                    res = self._distribute_output(name, res, last_inputs, to_run, waiting_for_input)
+                    for pair in self._find_components_that_received_no_input(name, res):
+                        _dequeue_component(pair, run_queue, waiting_queue)
+                    res = self._distribute_output(name, res, components_inputs, run_queue, waiting_queue)
 
                     if len(res) > 0:
                         final_outputs[name] = res
                 else:
                     # This component doesn't have enough inputs so we can't run it yet
-                    if (name, comp) not in waiting_for_input:
-                        waiting_for_input.append((name, comp))
+                    _enqueue_waiting_component((name, comp), waiting_queue)
 
-                if len(to_run) == 0 and len(waiting_for_input) > 0:
+                if len(run_queue) == 0 and len(waiting_queue) > 0:
                     # Check if we're stuck in a loop.
                     # It's important to check whether previous waitings are None as it could be that no
                     # Component has actually been run yet.
                     if (
-                        before_last_waiting_for_input is not None
-                        and last_waiting_for_input is not None
-                        and before_last_waiting_for_input == last_waiting_for_input
+                        before_last_waiting_queue is not None
+                        and last_waiting_queue is not None
+                        and before_last_waiting_queue == last_waiting_queue
                     ):
-                        # Are we actually stuck or there's a lazy variadic or a component with has only default inputs
-                        # waiting for input?
-                        # This is our last resort, if there's no lazy variadic or component with only default inputs
-                        # waiting for input we're stuck for real and we can't make any progress.
-                        for name, comp in waiting_for_input:
-                            is_variadic = any(
-                                socket.is_variadic
-                                for socket in comp.__haystack_input__._sockets_dict.values()  # type: ignore
+                        if self._is_stuck_in_a_loop(waiting_queue):
+                            # We're stuck! We can't make any progress.
+                            msg = (
+                                "Pipeline is stuck running in a loop. Partial outputs will be returned. "
+                                "Check the Pipeline graph for possible issues."
                             )
-                            has_only_defaults = all(
-                                not socket.is_mandatory
-                                for socket in comp.__haystack_input__._sockets_dict.values()  # type: ignore
-                            )
-                            if is_variadic and not comp.__haystack_is_greedy__ or has_only_defaults:  # type: ignore[attr-defined]
-                                break
-                        else:
-                            # We're stuck in a loop for real, we can't make any progress.
-                            # BAIL!
+                            warn(RuntimeWarning(msg))
                             break
 
-                        if len(waiting_for_input) == 1:
-                            # We have a single component with variadic input or only default inputs waiting for input.
-                            # If we're at this point it means it has been waiting for input for at least 2 iterations.
-                            # This will never run.
-                            # BAIL!
-                            break
-
-                        # There was a lazy variadic or a component with only default waiting for input, we can run it
-                        waiting_for_input.remove((name, comp))
-                        to_run.append((name, comp))
-
-                        # Let's use the default value for the inputs that are still missing, or the component
-                        # won't run and will be put back in the waiting list, causing an infinite loop.
-                        for input_socket in comp.__haystack_input__._sockets_dict.values():  # type: ignore
-                            if input_socket.is_mandatory:
-                                continue
-                            if input_socket.name not in last_inputs[name]:
-                                last_inputs[name][input_socket.name] = input_socket.default_value
-
+                        (name, comp) = self._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+                        _add_missing_input_defaults(name, comp, components_inputs)
+                        _enqueue_component((name, comp), run_queue, waiting_queue)
                         continue
 
-                    before_last_waiting_for_input = (
-                        last_waiting_for_input.copy() if last_waiting_for_input is not None else None
-                    )
-                    last_waiting_for_input = {item[0] for item in waiting_for_input}
+                    before_last_waiting_queue = last_waiting_queue.copy() if last_waiting_queue is not None else None
+                    last_waiting_queue = {item[0] for item in waiting_queue}
 
-                    self._enqueue_next_runnable_component(last_inputs, to_run, waiting_for_input)
+                    (name, comp) = self._find_next_runnable_component(components_inputs, waiting_queue)
+                    _add_missing_input_defaults(name, comp, components_inputs)
+                    _enqueue_component((name, comp), run_queue, waiting_queue)
 
             if len(include_outputs_from) > 0:
                 for name, output in extra_outputs.items():
diff --git a/releasenotes/notes/allow-parameters-for-openai-generator-8ce86ba1bac80817.yaml b/releasenotes/notes/allow-parameters-for-openai-generator-8ce86ba1bac80817.yaml
new file mode 100644
index 00000000..0c5e2b93
--- /dev/null
+++ b/releasenotes/notes/allow-parameters-for-openai-generator-8ce86ba1bac80817.yaml
@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    LLM based evaluators can pass in supported OpenAIGenerator parameters via api_params.
+    This allows for custom generation_kwargs, changing the api_base_url (for local evaluation),
+    and all other supported parameters as described in the OpenAIGenerator docs.
diff --git a/releasenotes/notes/improve-pypdf-backwards-compatibility-bcc75871005e9aba.yaml b/releasenotes/notes/improve-pypdf-backwards-compatibility-bcc75871005e9aba.yaml
deleted file mode 100644
index 48a46dfb..00000000
--- a/releasenotes/notes/improve-pypdf-backwards-compatibility-bcc75871005e9aba.yaml
+++ /dev/null
@@ -1,4 +0,0 @@
----
-enhancements:
- - |
-   Enhanced the PyPDF converter to ensure backwards compatibility with Pipelines dumped with versions older than 2.3.0. The update includes a conditional check to automatically default to the `DefaultConverter` if a specific converter is not provided, improving the component's robustness and ease of use.
diff --git a/releasenotes/notes/refactor-documentjoiner-992e662bee8ca219.yaml b/releasenotes/notes/refactor-documentjoiner-992e662bee8ca219.yaml
new file mode 100644
index 00000000..de31f8fe
--- /dev/null
+++ b/releasenotes/notes/refactor-documentjoiner-992e662bee8ca219.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Refactor DocumentJoiner to use enum pattern for the 'join_mode' parameter instead of bare string.


Test patch:
diff --git a/test/components/converters/test_pypdf_to_document.py b/test/components/converters/test_pypdf_to_document.py
index ecaba457..00530192 100644
--- a/test/components/converters/test_pypdf_to_document.py
+++ b/test/components/converters/test_pypdf_to_document.py
@@ -40,12 +40,6 @@ class TestPyPDFToDocument:
         assert isinstance(instance, PyPDFToDocument)
         assert isinstance(instance.converter, DefaultConverter)
 
-    def test_from_dict_no_converter(self):
-        data = {"type": "haystack.components.converters.pypdf.PyPDFToDocument", "init_parameters": {}}
-        instance = PyPDFToDocument.from_dict(data)
-        assert isinstance(instance, PyPDFToDocument)
-        assert isinstance(instance.converter, DefaultConverter)
-
     @pytest.mark.integration
     def test_run(self, test_files_path, pypdf_converter):
         """
diff --git a/test/components/evaluators/test_context_relevance_evaluator.py b/test/components/evaluators/test_context_relevance_evaluator.py
index 1a7daa0b..89f5b5ae 100644
--- a/test/components/evaluators/test_context_relevance_evaluator.py
+++ b/test/components/evaluators/test_context_relevance_evaluator.py
@@ -100,6 +100,7 @@ class TestContextRelevanceEvaluator:
             "init_parameters": {
                 "api_key": {"env_vars": ["ENV_VAR"], "strict": True, "type": "env_var"},
                 "api": "openai",
+                "api_params": {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}},
                 "examples": [{"inputs": {"questions": "What is football?"}, "outputs": {"score": 0}}],
                 "progress_bar": False,
                 "raise_on_failure": False,
diff --git a/test/components/evaluators/test_faithfulness_evaluator.py b/test/components/evaluators/test_faithfulness_evaluator.py
index 7616a884..8164ad9c 100644
--- a/test/components/evaluators/test_faithfulness_evaluator.py
+++ b/test/components/evaluators/test_faithfulness_evaluator.py
@@ -109,6 +109,7 @@ class TestFaithfulnessEvaluator:
             "init_parameters": {
                 "api_key": {"env_vars": ["ENV_VAR"], "strict": True, "type": "env_var"},
                 "api": "openai",
+                "api_params": {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}},
                 "examples": [
                     {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                 ],
diff --git a/test/components/evaluators/test_llm_evaluator.py b/test/components/evaluators/test_llm_evaluator.py
index edab8734..21d5fd02 100644
--- a/test/components/evaluators/test_llm_evaluator.py
+++ b/test/components/evaluators/test_llm_evaluator.py
@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
+import os
 from typing import List
 
 import pytest
@@ -23,6 +24,7 @@ class TestLLMEvaluator:
         )
         assert component.api == "openai"
         assert component.generator.client.api_key == "test-api-key"
+        assert component.api_params == {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}}
         assert component.instructions == "test-instruction"
         assert component.inputs == [("predicted_answers", List[str])]
         assert component.outputs == ["score"]
@@ -47,6 +49,7 @@ class TestLLMEvaluator:
         component = LLMEvaluator(
             instructions="test-instruction",
             api_key=Secret.from_token("test-api-key"),
+            api_params={"generation_kwargs": {"seed": 43}},
             inputs=[("predicted_answers", List[str])],
             outputs=["custom_score"],
             api="openai",
@@ -62,6 +65,7 @@ class TestLLMEvaluator:
             ],
         )
         assert component.generator.client.api_key == "test-api-key"
+        assert component.api_params == {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 43}}
         assert component.api == "openai"
         assert component.examples == [
             {"inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"}, "outputs": {"custom_score": 1}},
@@ -202,6 +206,7 @@ class TestLLMEvaluator:
         assert data == {
             "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
             "init_parameters": {
+                "api_params": {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}},
                 "api_key": {"env_vars": ["OPENAI_API_KEY"], "strict": True, "type": "env_var"},
                 "api": "openai",
                 "instructions": "test-instruction",
@@ -220,6 +225,7 @@ class TestLLMEvaluator:
         data = {
             "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
             "init_parameters": {
+                "api_params": {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}},
                 "api_key": {"env_vars": ["OPENAI_API_KEY"], "strict": True, "type": "env_var"},
                 "api": "openai",
                 "instructions": "test-instruction",
@@ -233,6 +239,7 @@ class TestLLMEvaluator:
         component = LLMEvaluator.from_dict(data)
         assert component.api == "openai"
         assert component.generator.client.api_key == "test-api-key"
+        assert component.api_params == {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}}
         assert component.instructions == "test-instruction"
         assert component.inputs == [("predicted_answers", List[str])]
         assert component.outputs == ["score"]
@@ -263,6 +270,7 @@ class TestLLMEvaluator:
         assert data == {
             "type": "haystack.components.evaluators.llm_evaluator.LLMEvaluator",
             "init_parameters": {
+                "api_params": {"generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42}},
                 "api_key": {"env_vars": ["ENV_VAR"], "strict": True, "type": "env_var"},
                 "api": "openai",
                 "instructions": "test-instruction",
@@ -442,3 +450,63 @@ class TestLLMEvaluator:
                     {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"score": 0}}
                 ],
             )
+
+    def test_init_with_base_url(self):
+        component = LLMEvaluator(
+            instructions="test-instruction",
+            api_key=Secret.from_token("test-api-key"),
+            api_params={"api_base_url": "http://127.0.0.1:11434/v1"},
+            inputs=[("predicted_answers", List[str])],
+            outputs=["custom_score"],
+            api="openai",
+            examples=[
+                {
+                    "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
+                    "outputs": {"custom_score": 1},
+                },
+                {
+                    "inputs": {"predicted_answers": "Football is the most popular sport."},
+                    "outputs": {"custom_score": 0},
+                },
+            ],
+        )
+        assert component.generator.client.api_key == "test-api-key"
+        assert component.api_params == {
+            "generation_kwargs": {"response_format": {"type": "json_object"}, "seed": 42},
+            "api_base_url": "http://127.0.0.1:11434/v1",
+        }
+        assert component.api == "openai"
+        assert component.examples == [
+            {"inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"}, "outputs": {"custom_score": 1}},
+            {"inputs": {"predicted_answers": "Football is the most popular sport."}, "outputs": {"custom_score": 0}},
+        ]
+        assert component.instructions == "test-instruction"
+        assert component.inputs == [("predicted_answers", List[str])]
+        assert component.outputs == ["custom_score"]
+
+    @pytest.mark.skipif(
+        not (os.environ.get("API_BASE_URL") and os.environ.get("MODEL_NAME")),
+        reason="Export env vars API_BASE_URL and MODEL_NAME containing the OpenAI API compatible server URL and the model name to run this test.",
+    )
+    @pytest.mark.integration
+    def test_run_with_base_url(self):
+        component = LLMEvaluator(
+            instructions="test-instruction",
+            api_key=Secret.from_token("test-api-key"),
+            api_params={"api_base_url": os.environ["API_BASE_URL"], "model": os.environ["MODEL_NAME"]},
+            inputs=[("predicted_answers", List[str])],
+            outputs=["custom_score"],
+            api="openai",
+            examples=[
+                {
+                    "inputs": {"predicted_answers": "Damn, this is straight outta hell!!!"},
+                    "outputs": {"custom_score": 1},
+                },
+                {
+                    "inputs": {"predicted_answers": "Football is the most popular sport."},
+                    "outputs": {"custom_score": 0},
+                },
+            ],
+        )
+        component.run(predicted_answers=["Damn, this is straight outta hell!!!", "Football is the most popular sport."])
+        assert component.outputs == ["custom_score"]
diff --git a/test/components/joiners/test_document_joiner.py b/test/components/joiners/test_document_joiner.py
index 0da08247..56bdfb64 100644
--- a/test/components/joiners/test_document_joiner.py
+++ b/test/components/joiners/test_document_joiner.py
@@ -2,28 +2,64 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 import logging
+import re
 
 import pytest
 
 from haystack import Document
-from haystack.components.joiners.document_joiner import DocumentJoiner
+from haystack.components.joiners.document_joiner import DocumentJoiner, JoinMode
 
 
 class TestDocumentJoiner:
     def test_init(self):
         joiner = DocumentJoiner()
-        assert joiner.join_mode == "concatenate"
+        assert joiner.join_mode == JoinMode.CONCATENATE
         assert joiner.weights is None
         assert joiner.top_k is None
         assert joiner.sort_by_score
 
     def test_init_with_custom_parameters(self):
         joiner = DocumentJoiner(join_mode="merge", weights=[0.4, 0.6], top_k=5, sort_by_score=False)
-        assert joiner.join_mode == "merge"
+        assert joiner.join_mode == JoinMode.MERGE
         assert joiner.weights == [0.4, 0.6]
         assert joiner.top_k == 5
         assert not joiner.sort_by_score
 
+    def test_to_dict(self):
+        joiner = DocumentJoiner()
+        data = joiner.to_dict()
+        assert data == {
+            "type": "haystack.components.joiners.document_joiner.DocumentJoiner",
+            "init_parameters": {"join_mode": "concatenate", "sort_by_score": True, "top_k": None, "weights": None},
+        }
+
+    def test_to_dict_custom_parameters(self):
+        joiner = DocumentJoiner("merge", weights=[0.4, 0.6], top_k=4, sort_by_score=False)
+        data = joiner.to_dict()
+        assert data == {
+            "type": "haystack.components.joiners.document_joiner.DocumentJoiner",
+            "init_parameters": {"join_mode": "merge", "weights": [0.4, 0.6], "top_k": 4, "sort_by_score": False},
+        }
+
+    def test_from_dict(self):
+        data = {"type": "haystack.components.joiners.document_joiner.DocumentJoiner", "init_parameters": {}}
+        document_joiner = DocumentJoiner.from_dict(data)
+        assert document_joiner.join_mode == JoinMode.CONCATENATE
+        assert document_joiner.weights == None
+        assert document_joiner.top_k == None
+        assert document_joiner.sort_by_score
+
+    def test_from_dict_customs_parameters(self):
+        data = {
+            "type": "haystack.components.joiners.document_joiner.DocumentJoiner",
+            "init_parameters": {"join_mode": "merge", "weights": [0.5, 0.6], "top_k": 6, "sort_by_score": False},
+        }
+        document_joiner = DocumentJoiner.from_dict(data)
+        assert document_joiner.join_mode == JoinMode.MERGE
+        assert document_joiner.weights == pytest.approx([0.5, 0.6], rel=0.1)
+        assert document_joiner.top_k == 6
+        assert not document_joiner.sort_by_score
+
     def test_empty_list(self):
         joiner = DocumentJoiner()
         result = joiner.run([])
@@ -41,8 +77,13 @@ class TestDocumentJoiner:
         assert result == {"documents": documents}
 
     def test_unsupported_join_mode(self):
-        with pytest.raises(ValueError, match="DocumentJoiner component does not support 'unsupported_mode' join_mode."):
-            DocumentJoiner(join_mode="unsupported_mode")
+        unsupported_mode = "unsupported_mode"
+        expected_error_pattern = (
+            re.escape(f"Unknown join mode '{unsupported_mode}'") + r".*Supported modes in DocumentJoiner are: \[.*\]"
+        )
+
+        with pytest.raises(ValueError, match=expected_error_pattern):
+            DocumentJoiner(join_mode=unsupported_mode)
 
     def test_run_with_concatenate_join_mode_and_top_k(self):
         joiner = DocumentJoiner(top_k=6)
diff --git a/test/core/pipeline/test_pipeline.py b/test/core/pipeline/test_pipeline.py
index 221c3523..9c38c6d0 100644
--- a/test/core/pipeline/test_pipeline.py
+++ b/test/core/pipeline/test_pipeline.py
@@ -8,12 +8,20 @@ from unittest.mock import patch
 import pytest
 
 from haystack import Document
-from haystack.components.builders import PromptBuilder
+from haystack.components.builders import PromptBuilder, AnswerBuilder
+from haystack.components.joiners import BranchJoiner
 from haystack.components.others import Multiplexer
 from haystack.core.component import component
 from haystack.core.component.types import InputSocket, OutputSocket, Variadic
 from haystack.core.errors import PipelineConnectError, PipelineDrawingError, PipelineError
 from haystack.core.pipeline import Pipeline, PredefinedPipeline
+from haystack.core.pipeline.base import (
+    _add_missing_input_defaults,
+    _enqueue_component,
+    _dequeue_component,
+    _enqueue_waiting_component,
+    _dequeue_waiting_component,
+)
 from haystack.core.serialization import DeserializationCallbacks
 from haystack.testing.factory import component_class
 from haystack.testing.sample_components import AddFixedValue, Double, Greet
@@ -765,7 +773,7 @@ class TestPipeline:
         for node in pipe.graph.nodes:
             assert pipe.graph.nodes[node]["visits"] == 0
 
-    def test__init_to_run(self):
+    def test__init_run_queue(self):
         ComponentWithVariadic = component_class(
             "ComponentWithVariadic", input_types={"in": Variadic[int]}, output_types={"out": int}
         )
@@ -792,13 +800,13 @@ class TestPipeline:
         pipe.connect("with_multiple_inputs.out", "with_variadic.in")
 
         data = {"yet_another_with_single_input": {"in": 1}}
-        to_run = pipe._init_to_run(data)
-        assert len(to_run) == 5
-        assert to_run[0][0] == "with_variadic"
-        assert to_run[1][0] == "with_no_inputs"
-        assert to_run[2][0] == "with_single_input"
-        assert to_run[3][0] == "yet_another_with_single_input"
-        assert to_run[4][0] == "with_multiple_inputs"
+        run_queue = pipe._init_run_queue(data)
+        assert len(run_queue) == 5
+        assert run_queue[0][0] == "with_variadic"
+        assert run_queue[1][0] == "with_no_inputs"
+        assert run_queue[2][0] == "with_single_input"
+        assert run_queue[3][0] == "yet_another_with_single_input"
+        assert run_queue[4][0] == "with_multiple_inputs"
 
     def test__init_inputs_state(self):
         pipe = Pipeline()
@@ -1089,7 +1097,7 @@ class TestPipeline:
             "sentence_builder", {"sentence_builder": {"words": ["blah blah"]}}
         )
 
-    def test__dequeue_components_that_received_no_input(self):
+    def test__find_components_that_received_no_input(self):
         sentence_builder = component_class(
             "SentenceBuilder", input_types={"words": List[str]}, output={"text": "some words"}
         )()
@@ -1102,11 +1110,11 @@ class TestPipeline:
         pipe.add_component("document_builder", document_builder)
         pipe.connect("sentence_builder.text", "document_builder.text")
 
-        to_run = [("document_builder", document_builder)]
-        waiting_for_input = [("document_builder", document_builder)]
-        pipe._dequeue_components_that_received_no_input("sentence_builder", {}, to_run, waiting_for_input)
-        assert to_run == []
-        assert waiting_for_input == []
+        res = pipe._find_components_that_received_no_input("sentence_builder", {})
+        assert res == {("document_builder", document_builder)}
+
+        res = pipe._find_components_that_received_no_input("sentence_builder", {"text": "some text"})
+        assert res == set()
 
     def test__distribute_output(self):
         document_builder = component_class(
@@ -1125,14 +1133,14 @@ class TestPipeline:
         pipe.connect("document_builder.another_doc", "document_joiner.docs")
 
         inputs = {"document_builder": {"text": "some text"}}
-        to_run = []
-        waiting_for_input = [("document_joiner", document_joiner)]
+        run_queue = []
+        waiting_queue = [("document_joiner", document_joiner)]
         res = pipe._distribute_output(
             "document_builder",
             {"doc": Document("some text"), "another_doc": Document()},
             inputs,
-            to_run,
-            waiting_for_input,
+            run_queue,
+            waiting_queue,
         )
 
         assert res == {}
@@ -1141,73 +1149,78 @@ class TestPipeline:
             "document_cleaner": {"doc": Document("some text")},
             "document_joiner": {"docs": [Document()]},
         }
-        assert to_run == [("document_cleaner", document_cleaner)]
-        assert waiting_for_input == [("document_joiner", document_joiner)]
+        assert run_queue == [("document_cleaner", document_cleaner)]
+        assert waiting_queue == [("document_joiner", document_joiner)]
 
-    def test__enqueue_next_runnable_component(self):
+    def test__find_next_runnable_component(self):
         document_builder = component_class(
             "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
         )()
         pipe = Pipeline()
-        inputs_by_component = {"document_builder": {"text": "some text"}}
-        to_run = []
-        waiting_for_input = [("document_builder", document_builder)]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
-
-        assert to_run == [("document_builder", document_builder)]
-        assert waiting_for_input == []
+        components_inputs = {"document_builder": {"text": "some text"}}
+        waiting_queue = [("document_builder", document_builder)]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
+        assert pair == ("document_builder", document_builder)
 
-    def test__enqueue_next_runnable_component_without_component_inputs(self):
+    def test__find_next_runnable_component_without_component_inputs(self):
         document_builder = component_class(
             "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
         )()
         pipe = Pipeline()
-        inputs_by_component = {}
-        to_run = []
-        waiting_for_input = [("document_builder", document_builder)]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
-
-        assert to_run == [("document_builder", document_builder)]
-        assert waiting_for_input == []
+        components_inputs = {}
+        waiting_queue = [("document_builder", document_builder)]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
+        assert pair == ("document_builder", document_builder)
 
-    def test__enqueue_next_runnable_component_with_component_with_only_variadic_non_greedy_input(self):
+    def test__find_next_runnable_component_with_component_with_only_variadic_non_greedy_input(self):
         document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
 
         pipe = Pipeline()
-        inputs_by_component = {}
-        to_run = []
-        waiting_for_input = [("document_joiner", document_joiner)]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
+        components_inputs = {}
+        waiting_queue = [("document_joiner", document_joiner)]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
+        assert pair == ("document_joiner", document_joiner)
+
+    def test__find_next_runnable_component_with_component_with_only_default_input(self):
+        prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
+
+        pipe = Pipeline()
+        components_inputs = {}
+        waiting_queue = [("prompt_builder", prompt_builder)]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
 
-        assert to_run == [("document_joiner", document_joiner)]
-        assert waiting_for_input == []
+        assert pair == ("prompt_builder", prompt_builder)
 
-    def test__enqueue_next_runnable_component_with_component_with_only_default_input(self):
+    def test__find_next_runnable_component_with_component_with_variadic_non_greedy_and_default_input(self):
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
         prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
 
         pipe = Pipeline()
-        inputs_by_component = {}
-        to_run = []
-        waiting_for_input = [("prompt_builder", prompt_builder)]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
+        components_inputs = {}
+        waiting_queue = [("prompt_builder", prompt_builder), ("document_joiner", document_joiner)]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
 
-        assert to_run == [("prompt_builder", prompt_builder)]
-        assert waiting_for_input == []
+        assert pair == ("document_joiner", document_joiner)
 
-    def test__enqueue_next_runnable_component_with_component_with_variadic_non_greedy_and_default_input(self):
+    def test__find_next_runnable_component_with_different_components_inputs(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
         document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
         prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
 
         pipe = Pipeline()
-        inputs_by_component = {}
-        to_run = []
-        waiting_for_input = [("prompt_builder", prompt_builder), ("document_joiner", document_joiner)]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
+        components_inputs = {"document_builder": {"text": "some text"}}
+        waiting_queue = [
+            ("prompt_builder", prompt_builder),
+            ("document_builder", document_builder),
+            ("document_joiner", document_joiner),
+        ]
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
 
-        assert to_run == [("document_joiner", document_joiner)]
-        assert waiting_for_input == [("prompt_builder", prompt_builder)]
+        assert pair == ("document_builder", document_builder)
 
-    def test__enqueue_next_runnable_component_with_different_components_inputs(self):
+    def test__find_next_runnable_component_with_different_components_without_any_input(self):
         document_builder = component_class(
             "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
         )()
@@ -1215,19 +1228,17 @@ class TestPipeline:
         prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
 
         pipe = Pipeline()
-        inputs_by_component = {"document_builder": {"text": "some text"}}
-        to_run = []
-        waiting_for_input = [
+        components_inputs = {}
+        waiting_queue = [
             ("prompt_builder", prompt_builder),
             ("document_builder", document_builder),
             ("document_joiner", document_joiner),
         ]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
+        pair = pipe._find_next_runnable_component(components_inputs, waiting_queue)
 
-        assert to_run == [("document_builder", document_builder)]
-        assert waiting_for_input == [("prompt_builder", prompt_builder), ("document_joiner", document_joiner)]
+        assert pair == ("document_builder", document_builder)
 
-    def test__enqueue_next_runnable_component_with_different_components_without_any_input(self):
+    def test__is_stuck_in_a_loop(self):
         document_builder = component_class(
             "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
         )()
@@ -1235,14 +1246,216 @@ class TestPipeline:
         prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
 
         pipe = Pipeline()
-        inputs_by_component = {}
-        to_run = []
-        waiting_for_input = [
+
+        waiting_queue = [("document_builder", document_builder)]
+        assert pipe._is_stuck_in_a_loop(waiting_queue)
+
+        waiting_queue = [("document_joiner", document_joiner)]
+        assert pipe._is_stuck_in_a_loop(waiting_queue)
+
+        waiting_queue = [("prompt_builder", prompt_builder)]
+        assert pipe._is_stuck_in_a_loop(waiting_queue)
+
+        waiting_queue = [("document_joiner", document_joiner), ("prompt_builder", prompt_builder)]
+        assert not pipe._is_stuck_in_a_loop(waiting_queue)
+
+        waiting_queue = [("document_builder", document_joiner), ("prompt_builder", prompt_builder)]
+        assert not pipe._is_stuck_in_a_loop(waiting_queue)
+
+        waiting_queue = [("document_builder", document_joiner), ("document_joiner", document_joiner)]
+        assert not pipe._is_stuck_in_a_loop(waiting_queue)
+
+    def test__enqueue_component(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
+
+        run_queue = []
+        waiting_queue = []
+        _enqueue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == [("document_builder", document_builder)]
+        assert waiting_queue == []
+
+        run_queue = [("document_builder", document_builder)]
+        waiting_queue = []
+        _enqueue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == [("document_builder", document_builder)]
+        assert waiting_queue == []
+
+        run_queue = []
+        waiting_queue = [("document_builder", document_builder)]
+        _enqueue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == [("document_builder", document_builder)]
+        assert waiting_queue == []
+
+        run_queue = []
+        waiting_queue = [("document_joiner", document_joiner)]
+        _enqueue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == [("document_builder", document_builder)]
+        assert waiting_queue == [("document_joiner", document_joiner)]
+
+        run_queue = [("document_joiner", document_joiner)]
+        waiting_queue = []
+        _enqueue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == [("document_joiner", document_joiner), ("document_builder", document_builder)]
+        assert waiting_queue == []
+
+    def test__dequeue_component(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
+
+        run_queue = []
+        waiting_queue = []
+        _dequeue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == []
+        assert waiting_queue == []
+
+        run_queue = [("document_builder", document_builder)]
+        waiting_queue = []
+        _dequeue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == []
+        assert waiting_queue == []
+
+        run_queue = []
+        waiting_queue = [("document_builder", document_builder)]
+        _dequeue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == []
+        assert waiting_queue == []
+
+        run_queue = [("document_builder", document_builder)]
+        waiting_queue = [("document_builder", document_builder)]
+        _dequeue_component(("document_builder", document_builder), run_queue, waiting_queue)
+        assert run_queue == []
+        assert waiting_queue == []
+
+        run_queue = [("document_builder", document_builder)]
+        waiting_queue = [("document_builder", document_builder)]
+        _dequeue_component(("document_joiner", document_joiner), run_queue, waiting_queue)
+        assert run_queue == [("document_builder", document_builder)]
+        assert waiting_queue == [("document_builder", document_builder)]
+
+    def test__add_missing_input_defaults(self):
+        name = "prompt_builder"
+        prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
+        components_inputs = {}
+        _add_missing_input_defaults(name, prompt_builder, components_inputs)
+        assert components_inputs == {"prompt_builder": {"questions": "", "template": None, "template_variables": None}}
+
+        name = "answer_builder"
+        answer_builder = AnswerBuilder()
+        components_inputs = {"answer_builder": {"query": "What is the answer?"}}
+        _add_missing_input_defaults(name, answer_builder, components_inputs)
+        assert components_inputs == {
+            "answer_builder": {
+                "query": "What is the answer?",
+                "meta": None,
+                "documents": None,
+                "pattern": None,
+                "reference_pattern": None,
+            }
+        }
+
+        name = "branch_joiner"
+        branch_joiner = BranchJoiner(int)
+        components_inputs = {}
+        _add_missing_input_defaults(name, branch_joiner, components_inputs)
+        assert components_inputs == {"branch_joiner": {}}
+
+    def test__find_next_runnable_lazy_variadic_or_default_component(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
+        prompt_builder = PromptBuilder(template="{{ questions | join('\n') }}")
+        pipe = Pipeline()
+
+        waiting_queue = [("document_builder", document_builder)]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("document_builder", document_builder)
+
+        waiting_queue = [("document_joiner", document_joiner)]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("document_joiner", document_joiner)
+
+        waiting_queue = [("prompt_builder", prompt_builder)]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("prompt_builder", prompt_builder)
+
+        waiting_queue = [
+            ("document_builder", document_builder),
+            ("document_joiner", document_joiner),
+            ("prompt_builder", prompt_builder),
+        ]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("document_joiner", document_joiner)
+
+        waiting_queue = [
             ("prompt_builder", prompt_builder),
             ("document_builder", document_builder),
             ("document_joiner", document_joiner),
         ]
-        pipe._enqueue_next_runnable_component(inputs_by_component, to_run, waiting_for_input)
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("prompt_builder", prompt_builder)
+
+        waiting_queue = [
+            ("document_builder", document_builder),
+            ("document_joiner", document_joiner),
+            ("prompt_builder", prompt_builder),
+        ]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("document_joiner", document_joiner)
+
+        waiting_queue = [
+            ("document_builder", document_builder),
+            ("prompt_builder", prompt_builder),
+            ("document_joiner", document_joiner),
+        ]
+        pair = pipe._find_next_runnable_lazy_variadic_or_default_component(waiting_queue)
+        assert pair == ("prompt_builder", prompt_builder)
+
+    def test__enqueue_waiting_component(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
+
+        waiting_queue = []
+        _enqueue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_builder", document_builder)]
+
+        waiting_queue = [("document_builder", document_builder)]
+        _enqueue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_builder", document_builder)]
+
+        waiting_queue = [("document_joiner", document_joiner)]
+        _enqueue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_joiner", document_joiner), ("document_builder", document_builder)]
+
+        waiting_queue = [("document_builder", document_builder), ("document_joiner", document_joiner)]
+        _enqueue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_builder", document_builder), ("document_joiner", document_joiner)]
+
+    def test__dequeue_waiting_component(self):
+        document_builder = component_class(
+            "DocumentBuilder", input_types={"text": str}, output_types={"doc": Document}
+        )()
+        document_joiner = component_class("DocumentJoiner", input_types={"docs": Variadic[Document]})()
+
+        waiting_queue = []
+        _dequeue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == []
+
+        waiting_queue = [("document_builder", document_builder)]
+        _dequeue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == []
+
+        waiting_queue = [("document_joiner", document_joiner)]
+        _dequeue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_joiner", document_joiner)]
 
-        assert to_run == [("document_builder", document_builder)]
-        assert waiting_for_input == [("prompt_builder", prompt_builder), ("document_joiner", document_joiner)]
+        waiting_queue = [("document_builder", document_builder), ("document_joiner", document_joiner)]
+        _dequeue_waiting_component(("document_builder", document_builder), waiting_queue)
+        assert waiting_queue == [("document_joiner", document_joiner)]