Repository: deepset-ai/haystack

Issue:
Merge `hf_utils.py` files
`hf_utils.py` exists both in [`haystack.components.generators`](https://github.com/deepset-ai/haystack/blob/main/haystack/components/generators/hf_utils.py) and [`haystack.components.embedders`](https://github.com/deepset-ai/haystack/blob/main/haystack/components/embedders/hf_utils.py).

We must merged in a single file, if necessary move it in a more sensible package.




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


* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **36** unchanged lines in **2** files lost coverage.
* Overall coverage increased (+**0.05%**) to **88.287%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/generators/chat/hugging_face_local.py](https://coveralls.io/builds/65506662/source?filename=components%2Fgenerators%2Fchat%2Fhugging_face_local.py#L143) | 12 | 86.49% |
| [utils/hf.py](https://coveralls.io/builds/65506662/source?filename=utils%2Fhf.py#L105) | 24 | 76.92% |
<!-- | **Total:** | **36** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/65506662/badge)](https://coveralls.io/builds/65506662) |
| :-- | --: |
| Change from base [Build 7786748820](https://coveralls.io/builds/65504683): |  0.05% |
| Covered Lines: | 4696 |
| Relevant Lines: | 5319 |

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


Patch:
diff --git a/haystack/components/embedders/hf_utils.py b/haystack/components/embedders/hf_utils.py
deleted file mode 100644
index 81ec246d..00000000
--- a/haystack/components/embedders/hf_utils.py
+++ /dev/null
@@ -1,32 +0,0 @@
-from typing import Optional
-
-from haystack.lazy_imports import LazyImport
-from haystack.utils.auth import Secret
-
-with LazyImport(message="Run 'pip install transformers'") as transformers_import:
-    from huggingface_hub import HfApi
-    from huggingface_hub.utils import RepositoryNotFoundError
-
-
-def check_valid_model(model_id: str, token: Optional[Secret]) -> None:
-    """
-    Check if the provided model ID corresponds to a valid model on HuggingFace Hub.
-    Also check if the model is a embedding model.
-
-    :param model_id: A string representing the HuggingFace model ID.
-    :param token: The optional authentication token.
-    :raises ValueError: If the model is not found or is not a embedding model.
-    """
-    transformers_import.check()
-
-    api = HfApi()
-    try:
-        model_info = api.model_info(model_id, token=token.resolve_value() if token else None)
-    except RepositoryNotFoundError as e:
-        raise ValueError(
-            f"Model {model_id} not found on HuggingFace Hub. Please provide a valid HuggingFace model_id."
-        ) from e
-
-    allowed_model = model_info.pipeline_tag in ["sentence-similarity", "feature-extraction"]
-    if not allowed_model:
-        raise ValueError(f"Model {model_id} is not a embedding model. Please provide a embedding model.")
diff --git a/haystack/components/embedders/hugging_face_tei_document_embedder.py b/haystack/components/embedders/hugging_face_tei_document_embedder.py
index e03ff4bf..38123f04 100644
--- a/haystack/components/embedders/hugging_face_tei_document_embedder.py
+++ b/haystack/components/embedders/hugging_face_tei_document_embedder.py
@@ -4,10 +4,10 @@ from urllib.parse import urlparse
 
 from tqdm import tqdm
 
-from haystack.components.embedders.hf_utils import check_valid_model
 from haystack.dataclasses import Document
 from haystack.lazy_imports import LazyImport
 from haystack.utils import Secret, deserialize_secrets_inplace
+from haystack.utils.hf import check_valid_model, HFModelType
 from haystack import component, default_to_dict, default_from_dict
 
 with LazyImport(message="Run 'pip install transformers'") as transformers_import:
@@ -116,7 +116,7 @@ class HuggingFaceTEIDocumentEmbedder:
             if not is_valid_url:
                 raise ValueError(f"Invalid TEI endpoint URL provided: {url}")
 
-        check_valid_model(model, token)
+        check_valid_model(model, HFModelType.EMBEDDING, token)
 
         self.model = model
         self.url = url
diff --git a/haystack/components/embedders/hugging_face_tei_text_embedder.py b/haystack/components/embedders/hugging_face_tei_text_embedder.py
index 0acedbce..339454b1 100644
--- a/haystack/components/embedders/hugging_face_tei_text_embedder.py
+++ b/haystack/components/embedders/hugging_face_tei_text_embedder.py
@@ -3,9 +3,9 @@ from typing import Any, Dict, List, Optional
 from urllib.parse import urlparse
 
 from haystack import component, default_to_dict, default_from_dict
-from haystack.components.embedders.hf_utils import check_valid_model
 from haystack.lazy_imports import LazyImport
 from haystack.utils import Secret, deserialize_secrets_inplace
+from haystack.utils.hf import check_valid_model, HFModelType
 
 with LazyImport(message="Run 'pip install transformers'") as transformers_import:
     from huggingface_hub import InferenceClient
@@ -98,7 +98,7 @@ class HuggingFaceTEITextEmbedder:
             if not is_valid_url:
                 raise ValueError(f"Invalid TEI endpoint URL provided: {url}")
 
-        check_valid_model(model, token)
+        check_valid_model(model, HFModelType.EMBEDDING, token)
 
         self.model = model
         self.url = url
diff --git a/haystack/components/generators/chat/hugging_face_local.py b/haystack/components/generators/chat/hugging_face_local.py
index 7ccde0e8..50115427 100644
--- a/haystack/components/generators/chat/hugging_face_local.py
+++ b/haystack/components/generators/chat/hugging_face_local.py
@@ -2,8 +2,6 @@ import logging
 import sys
 from typing import Any, Dict, List, Literal, Optional, Union, Callable
 
-from haystack.components.generators.hf_utils import PIPELINE_SUPPORTED_TASKS
-
 from haystack import component, default_to_dict, default_from_dict
 from haystack.components.generators.utils import serialize_callback_handler, deserialize_callback_handler
 from haystack.dataclasses import ChatMessage, StreamingChunk
@@ -16,11 +14,15 @@ logger = logging.getLogger(__name__)
 with LazyImport(message="Run 'pip install transformers[torch]'") as torch_and_transformers_import:
     from huggingface_hub import model_info
     from transformers import StoppingCriteriaList, pipeline, PreTrainedTokenizer, PreTrainedTokenizerFast
-    from haystack.components.generators.hf_utils import (  # pylint: disable=ungrouped-imports
+    from haystack.utils.hf import (  # pylint: disable=ungrouped-imports
         StopWordsCriteria,
         HFTokenStreamingHandler,
+        serialize_hf_model_kwargs,
+        deserialize_hf_model_kwargs,
     )
-    from haystack.utils.hf import serialize_hf_model_kwargs, deserialize_hf_model_kwargs
+
+
+PIPELINE_SUPPORTED_TASKS = ["text-generation", "text2text-generation"]
 
 
 @component
diff --git a/haystack/components/generators/chat/hugging_face_tgi.py b/haystack/components/generators/chat/hugging_face_tgi.py
index 1ff241c5..17462162 100644
--- a/haystack/components/generators/chat/hugging_face_tgi.py
+++ b/haystack/components/generators/chat/hugging_face_tgi.py
@@ -6,9 +6,9 @@ from urllib.parse import urlparse
 from haystack import component, default_to_dict, default_from_dict
 from haystack.components.generators.utils import serialize_callback_handler, deserialize_callback_handler
 from haystack.dataclasses import ChatMessage, StreamingChunk
-from haystack.components.generators.hf_utils import check_valid_model, check_generation_params
 from haystack.lazy_imports import LazyImport
 from haystack.utils import Secret, deserialize_secrets_inplace
+from haystack.utils.hf import check_valid_model, HFModelType, check_generation_params
 
 with LazyImport(message="Run 'pip install transformers'") as transformers_import:
     from huggingface_hub import InferenceClient
@@ -120,7 +120,7 @@ class HuggingFaceTGIChatGenerator:
             if not is_valid_url:
                 raise ValueError(f"Invalid TGI endpoint URL provided: {url}")
 
-        check_valid_model(model, token)
+        check_valid_model(model, HFModelType.GENERATION, token)
 
         # handle generation kwargs setup
         generation_kwargs = generation_kwargs.copy() if generation_kwargs else {}
diff --git a/haystack/components/generators/hf_utils.py b/haystack/components/generators/hf_utils.py
deleted file mode 100644
index dea89372..00000000
--- a/haystack/components/generators/hf_utils.py
+++ /dev/null
@@ -1,131 +0,0 @@
-import inspect
-from typing import Any, Dict, List, Optional, Union, Callable
-
-from haystack.dataclasses import StreamingChunk
-from haystack.lazy_imports import LazyImport
-from haystack.utils import Secret
-
-with LazyImport(message="Run 'pip install transformers'") as transformers_import:
-    from huggingface_hub import InferenceClient, HfApi
-    from huggingface_hub.utils import RepositoryNotFoundError
-
-PIPELINE_SUPPORTED_TASKS = ["text-generation", "text2text-generation"]
-
-
-def check_generation_params(kwargs: Optional[Dict[str, Any]], additional_accepted_params: Optional[List[str]] = None):
-    """
-    Check the provided generation parameters for validity.
-
-    :param kwargs: A dictionary containing the generation parameters.
-    :param additional_accepted_params: An optional list of strings representing additional accepted parameters.
-    :raises ValueError: If any unknown text generation parameters are provided.
-    """
-    transformers_import.check()
-
-    if kwargs:
-        accepted_params = {
-            param
-            for param in inspect.signature(InferenceClient.text_generation).parameters.keys()
-            if param not in ["self", "prompt"]
-        }
-        if additional_accepted_params:
-            accepted_params.update(additional_accepted_params)
-        unknown_params = set(kwargs.keys()) - accepted_params
-        if unknown_params:
-            raise ValueError(
-                f"Unknown text generation parameters: {unknown_params}. The valid parameters are: {accepted_params}."
-            )
-
-
-def check_valid_model(model_id: str, token: Optional[Secret]) -> None:
-    """
-    Check if the provided model ID corresponds to a valid model on HuggingFace Hub.
-    Also check if the model is a text generation model.
-
-    :param model_id: A string representing the HuggingFace model ID.
-    :param token: An optional authentication token.
-    :raises ValueError: If the model is not found or is not a text generation model.
-    """
-    transformers_import.check()
-
-    api = HfApi()
-    try:
-        model_info = api.model_info(model_id, token=token.resolve_value() if token else None)
-    except RepositoryNotFoundError as e:
-        raise ValueError(
-            f"Model {model_id} not found on HuggingFace Hub. Please provide a valid HuggingFace model_id."
-        ) from e
-
-    allowed_model = model_info.pipeline_tag in ["text-generation", "text2text-generation"]
-    if not allowed_model:
-        raise ValueError(f"Model {model_id} is not a text generation model. Please provide a text generation model.")
-
-
-with LazyImport(message="Run 'pip install transformers[torch]'") as torch_and_transformers_import:
-    import torch
-    from transformers import StoppingCriteria, PreTrainedTokenizer, PreTrainedTokenizerFast, TextStreamer
-
-    transformers_import.check()
-
-    class StopWordsCriteria(StoppingCriteria):
-        """
-        Stops text generation if any one of the stop words is generated.
-
-        Note: When a stop word is encountered, the generation of new text is stopped.
-        However, if the stop word is in the prompt itself, it can stop generating new text
-        prematurely after the first token. This is particularly important for LLMs designed
-        for dialogue generation. For these models, like for example mosaicml/mpt-7b-chat,
-        the output includes both the new text and the original prompt. Therefore, it's important
-        to make sure your prompt has no stop words.
-        """
-
-        def __init__(
-            self,
-            tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
-            stop_words: List[str],
-            device: Union[str, torch.device] = "cpu",
-        ):
-            super().__init__()
-            # check if tokenizer is a valid tokenizer
-            if not isinstance(tokenizer, (PreTrainedTokenizer, PreTrainedTokenizerFast)):
-                raise ValueError(
-                    f"Invalid tokenizer provided for StopWordsCriteria - {tokenizer}. "
-                    f"Please provide a valid tokenizer from the HuggingFace Transformers library."
-                )
-            if not tokenizer.pad_token:
-                if tokenizer.eos_token:
-                    tokenizer.pad_token = tokenizer.eos_token
-                else:
-                    tokenizer.add_special_tokens({"pad_token": "[PAD]"})
-            encoded_stop_words = tokenizer(stop_words, add_special_tokens=False, padding=True, return_tensors="pt")
-            self.stop_ids = encoded_stop_words.input_ids.to(device)
-
-        def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool:
-            for stop_id in self.stop_ids:
-                found_stop_word = self.is_stop_word_found(input_ids, stop_id)
-                if found_stop_word:
-                    return True
-            return False
-
-        def is_stop_word_found(self, generated_text_ids: torch.Tensor, stop_id: torch.Tensor) -> bool:
-            generated_text_ids = generated_text_ids[-1]
-            len_generated_text_ids = generated_text_ids.size(0)
-            len_stop_id = stop_id.size(0)
-            result = all(generated_text_ids[len_generated_text_ids - len_stop_id :].eq(stop_id))
-            return result
-
-    class HFTokenStreamingHandler(TextStreamer):
-        def __init__(
-            self,
-            tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
-            stream_handler: Callable[[StreamingChunk], None],
-            stop_words: Optional[List[str]] = None,
-        ):
-            super().__init__(tokenizer=tokenizer, skip_prompt=True)  # type: ignore
-            self.token_handler = stream_handler
-            self.stop_words = stop_words or []
-
-        def on_finalized_text(self, word: str, stream_end: bool = False):
-            word_to_send = word + "\n" if stream_end else word
-            if word_to_send.strip() not in self.stop_words:
-                self.token_handler(StreamingChunk(content=word_to_send))
diff --git a/haystack/components/generators/hugging_face_local.py b/haystack/components/generators/hugging_face_local.py
index 4a7eb7b6..a6286a92 100644
--- a/haystack/components/generators/hugging_face_local.py
+++ b/haystack/components/generators/hugging_face_local.py
@@ -15,7 +15,7 @@ SUPPORTED_TASKS = ["text-generation", "text2text-generation"]
 with LazyImport(message="Run 'pip install transformers[torch]'") as transformers_import:
     from huggingface_hub import model_info
     from transformers import StoppingCriteriaList, pipeline
-    from haystack.components.generators.hf_utils import StopWordsCriteria  # pylint: disable=ungrouped-imports
+    from haystack.utils.hf import StopWordsCriteria  # pylint: disable=ungrouped-imports
 
 
 @component
diff --git a/haystack/components/generators/hugging_face_tgi.py b/haystack/components/generators/hugging_face_tgi.py
index 0485a011..d7499d58 100644
--- a/haystack/components/generators/hugging_face_tgi.py
+++ b/haystack/components/generators/hugging_face_tgi.py
@@ -6,9 +6,9 @@ from urllib.parse import urlparse
 from haystack import component, default_to_dict, default_from_dict
 from haystack.components.generators.utils import serialize_callback_handler, deserialize_callback_handler
 from haystack.dataclasses import StreamingChunk
-from haystack.components.generators.hf_utils import check_generation_params, check_valid_model
 from haystack.lazy_imports import LazyImport
 from haystack.utils import Secret, deserialize_secrets_inplace
+from haystack.utils.hf import check_valid_model, HFModelType, check_generation_params
 
 with LazyImport(message="Run 'pip install transformers'") as transformers_import:
     from huggingface_hub import InferenceClient
@@ -104,7 +104,7 @@ class HuggingFaceTGIGenerator:
             if not is_valid_url:
                 raise ValueError(f"Invalid TGI endpoint URL provided: {url}")
 
-        check_valid_model(model, token)
+        check_valid_model(model, HFModelType.GENERATION, token)
 
         # handle generation kwargs setup
         generation_kwargs = generation_kwargs.copy() if generation_kwargs else {}
diff --git a/haystack/core/component/__init__.py b/haystack/core/component/__init__.py
index 3a292eda..dea76141 100644
--- a/haystack/core/component/__init__.py
+++ b/haystack/core/component/__init__.py
@@ -1,7 +1,7 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
-from haystack.core.component.component import component, Component
-from haystack.core.component.sockets import InputSocket, OutputSocket
+from haystack.core.component.component import Component, component
+from haystack.core.component.types import InputSocket, OutputSocket
 
 __all__ = ["component", "Component", "InputSocket", "OutputSocket"]
diff --git a/haystack/core/component/component.py b/haystack/core/component/component.py
index 5fab8455..95ae87d7 100644
--- a/haystack/core/component/component.py
+++ b/haystack/core/component/component.py
@@ -74,9 +74,11 @@ from copy import deepcopy
 from types import new_class
 from typing import Any, Protocol, runtime_checkable
 
-from haystack.core.component.sockets import InputSocket, OutputSocket, _empty
 from haystack.core.errors import ComponentError
 
+from .sockets import Sockets
+from .types import InputSocket, OutputSocket, _empty
+
 logger = logging.getLogger(__name__)
 
 
@@ -131,12 +133,14 @@ class ComponentMeta(type):
             # that stores the output specification.
             # We deepcopy the content of the cache to transfer ownership from the class method
             # to the actual instance, so that different instances of the same class won't share this data.
-            instance.__haystack_output__ = deepcopy(getattr(instance.run, "_output_types_cache", {}))
+            instance.__haystack_output__ = Sockets(
+                instance, deepcopy(getattr(instance.run, "_output_types_cache", {})), OutputSocket
+            )
 
         # Create the sockets if set_input_types() wasn't called in the constructor.
         # If it was called and there are some parameters also in the `run()` method, these take precedence.
         if not hasattr(instance, "__haystack_input__"):
-            instance.__haystack_input__ = {}
+            instance.__haystack_input__ = Sockets(instance, {}, InputSocket)
         run_signature = inspect.signature(getattr(cls, "run"))
         for param in list(run_signature.parameters)[1:]:  # First is 'self' and it doesn't matter.
             if run_signature.parameters[param].kind not in (
@@ -185,7 +189,7 @@ class _Component:
         :param default: default value of the input socket, defaults to _empty
         """
         if not hasattr(instance, "__haystack_input__"):
-            instance.__haystack_input__ = {}
+            instance.__haystack_input__ = Sockets(instance, {}, InputSocket)
         instance.__haystack_input__[name] = InputSocket(name=name, type=type, default_value=default)
 
     def set_input_types(self, instance, **types):
@@ -229,7 +233,9 @@ class _Component:
         parameter mandatory as specified in `set_input_types`.
 
         """
-        instance.__haystack_input__ = {name: InputSocket(name=name, type=type_) for name, type_ in types.items()}
+        instance.__haystack_input__ = Sockets(
+            instance, {name: InputSocket(name=name, type=type_) for name, type_ in types.items()}, InputSocket
+        )
 
     def set_output_types(self, instance, **types):
         """
@@ -251,7 +257,9 @@ class _Component:
                 return {"output_1": 1, "output_2": "2"}
         ```
         """
-        instance.__haystack_output__ = {name: OutputSocket(name=name, type=type_) for name, type_ in types.items()}
+        instance.__haystack_output__ = Sockets(
+            instance, {name: OutputSocket(name=name, type=type_) for name, type_ in types.items()}, OutputSocket
+        )
 
     def output_types(self, **types):
         """
diff --git a/haystack/core/component/sockets.py b/haystack/core/component/sockets.py
index ff3080dc..25bf4fdc 100644
--- a/haystack/core/component/sockets.py
+++ b/haystack/core/component/sockets.py
@@ -1,57 +1,107 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
+
 import logging
-from dataclasses import dataclass, field
-from typing import Any, List, Type, get_args
+from typing import Dict, Type, Union
+
+from haystack.core.type_utils import _type_name
 
-from haystack.core.component.types import HAYSTACK_VARIADIC_ANNOTATION
+from .types import InputSocket, OutputSocket
 
 logger = logging.getLogger(__name__)
 
+SocketsDict = Dict[str, Union[InputSocket, OutputSocket]]
+SocketsIOType = Union[Type[InputSocket], Type[OutputSocket]]
+
+
+class Sockets:
+    """
+    This class is used to represent the inputs or outputs of a `Component`.
+    Depending on the type passed to the constructor, it will represent either the inputs or the outputs of
+    the `Component`.
+
+    Usage:
+    ```python
+    from haystack.components.builders.prompt_builder import PromptBuilder
+
+    prompt_template = \"""
+    Given these documents, answer the question.\nDocuments:
+    {% for doc in documents %}
+        {{ doc.content }}
+    {% endfor %}
+
+    \nQuestion: {{question}}
+    \nAnswer:
+    \"""
 
-class _empty:
-    """Custom object for marking InputSocket.default_value as not set."""
+    prompt_builder = PromptBuilder(template=prompt_template)
+    sockets = {"question": InputSocket("question", Any), "documents": InputSocket("documents", Any)}
+    inputs = Sockets(component=prompt_builder, sockets=sockets, sockets_type=InputSocket)
+    inputs
+    >>> PromptBuilder inputs:
+    >>>   - question: Any
+    >>>   - documents: Any
 
+    inputs.question
+    >>> InputSocket(name='question', type=typing.Any, default_value=<class 'haystack.core.component.types._empty'>, is_variadic=False, senders=[])
+    ```
+    """
 
-@dataclass
-class InputSocket:
-    name: str
-    type: Type
-    default_value: Any = _empty
-    is_variadic: bool = field(init=False)
-    senders: List[str] = field(default_factory=list)
+    # We're using a forward declaration here to avoid a circular import.
+    def __init__(
+        self,
+        component: "Component",  # type: ignore[name-defined] # noqa: F821
+        sockets_dict: SocketsDict,
+        sockets_io_type: SocketsIOType,
+    ):
+        """
+        Create a new Sockets object.
+        We don't do any enforcement on the types of the sockets here, the `sockets_type` is only used for
+        the `__repr__` method.
+        We could do without it and use the type of a random value in the `sockets` dict, but that wouldn't
+        work for components that have no sockets at all. Either input or output.
+        """
+        self._sockets_io_type = sockets_io_type
+        self._component = component
+        self._sockets_dict = sockets_dict
+        self.__dict__.update(sockets_dict)
 
-    @property
-    def is_mandatory(self):
-        return self.default_value == _empty
+    def __setitem__(self, key: str, socket: Union[InputSocket, OutputSocket]):
+        """
+        Adds a new socket to this Sockets object.
+        This eases a bit updating the list of sockets after Sockets has been created.
+        That should happen only in the `component` decorator.
+        """
+        self._sockets_dict[key] = socket
+        self.__dict__[key] = socket
 
-    def __post_init__(self):
+    def _component_name(self) -> str:
+        if pipeline := getattr(self._component, "__haystack_added_to_pipeline__"):
+            # This Component has been added in a Pipeline, let's get the name from there.
+            return pipeline.get_component_name(self._component)
+
+        # This Component has not been added to a Pipeline yet, so we can't know its name.
+        # Let's use the class name instead.
+        return str(self._component)
+
+    def __getattribute__(self, name):
         try:
-            # __metadata__ is a tuple
-            self.is_variadic = self.type.__metadata__[0] == HAYSTACK_VARIADIC_ANNOTATION
+            sockets = object.__getattribute__(self, "_sockets")
+            if name in sockets:
+                return sockets[name]
         except AttributeError:
-            self.is_variadic = False
-        if self.is_variadic:
-            # We need to "unpack" the type inside the Variadic annotation,
-            # otherwise the pipeline connection api will try to match
-            # `Annotated[type, CANALS_VARIADIC_ANNOTATION]`.
-            #
-            # Note1: Variadic is expressed as an annotation of one single type,
-            # so the return value of get_args will always be a one-item tuple.
-            #
-            # Note2: a pipeline always passes a list of items when a component
-            # input is declared as Variadic, so the type itself always wraps
-            # an iterable of the declared type. For example, Variadic[int]
-            # is eventually an alias for Iterable[int]. Since we're interested
-            # in getting the inner type `int`, we call `get_args` twice: the
-            # first time to get `List[int]` out of `Variadic`, the second time
-            # to get `int` out of `List[int]`.
-            self.type = get_args(get_args(self.type)[0])[0]
-
-
-@dataclass
-class OutputSocket:
-    name: str
-    type: type
-    receivers: List[str] = field(default_factory=list)
+            pass
+
+        return object.__getattribute__(self, name)
+
+    def __repr__(self) -> str:
+        result = self._component_name()
+        if self._sockets_io_type == InputSocket:
+            result += " inputs:\n"
+        elif self._sockets_io_type == OutputSocket:
+            result += " outputs:\n"
+
+        result += "\n".join([f"  - {n}: {_type_name(s.type)}" for n, s in self._sockets_dict.items()])
+
+        return result
diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py
index 25278790..663bccf4 100644
--- a/haystack/core/component/types.py
+++ b/haystack/core/component/types.py
@@ -1,4 +1,5 @@
-from typing import Iterable, TypeVar
+from dataclasses import dataclass, field
+from typing import Any, Iterable, List, Type, TypeVar, get_args
 
 from typing_extensions import Annotated, TypeAlias  # Python 3.8 compatibility
 
@@ -13,3 +14,50 @@ T = TypeVar("T")
 # type so it can be used in the `InputSocket` creation where we
 # check that its annotation equals to CANALS_VARIADIC_ANNOTATION
 Variadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_VARIADIC_ANNOTATION]
+
+
+class _empty:
+    """Custom object for marking InputSocket.default_value as not set."""
+
+
+@dataclass
+class InputSocket:
+    name: str
+    type: Type
+    default_value: Any = _empty
+    is_variadic: bool = field(init=False)
+    senders: List[str] = field(default_factory=list)
+
+    @property
+    def is_mandatory(self):
+        return self.default_value == _empty
+
+    def __post_init__(self):
+        try:
+            # __metadata__ is a tuple
+            self.is_variadic = self.type.__metadata__[0] == HAYSTACK_VARIADIC_ANNOTATION
+        except AttributeError:
+            self.is_variadic = False
+        if self.is_variadic:
+            # We need to "unpack" the type inside the Variadic annotation,
+            # otherwise the pipeline connection api will try to match
+            # `Annotated[type, HAYSTACK_VARIADIC_ANNOTATION]`.
+            #
+            # Note1: Variadic is expressed as an annotation of one single type,
+            # so the return value of get_args will always be a one-item tuple.
+            #
+            # Note2: a pipeline always passes a list of items when a component
+            # input is declared as Variadic, so the type itself always wraps
+            # an iterable of the declared type. For example, Variadic[int]
+            # is eventually an alias for Iterable[int]. Since we're interested
+            # in getting the inner type `int`, we call `get_args` twice: the
+            # first time to get `List[int]` out of `Variadic`, the second time
+            # to get `int` out of `List[int]`.
+            self.type = get_args(get_args(self.type)[0])[0]
+
+
+@dataclass
+class OutputSocket:
+    name: str
+    type: type
+    receivers: List[str] = field(default_factory=list)
diff --git a/haystack/core/pipeline/descriptions.py b/haystack/core/pipeline/descriptions.py
index 406bf196..0e7c0420 100644
--- a/haystack/core/pipeline/descriptions.py
+++ b/haystack/core/pipeline/descriptions.py
@@ -1,14 +1,13 @@
 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
 #
 # SPDX-License-Identifier: Apache-2.0
-from typing import List, Dict
 import logging
+from typing import Dict, List
 
 import networkx  # type:ignore
 
+from haystack.core.component.types import InputSocket, OutputSocket
 from haystack.core.type_utils import _type_name
-from haystack.core.component.sockets import InputSocket, OutputSocket
-
 
 logger = logging.getLogger(__name__)
 
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index eeecdcea..a435f520 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -197,16 +197,17 @@ class Pipeline:
             )
             raise PipelineError(msg)
 
-        # Create the component's input and output sockets
-        input_sockets = getattr(instance, "__haystack_input__", {})
-        output_sockets = getattr(instance, "__haystack_output__", {})
-
         setattr(instance, "__haystack_added_to_pipeline__", self)
 
         # Add component to the graph, disconnected
         logger.debug("Adding component '%s' (%s)", name, instance)
+        # We're completely sure the fields exist so we ignore the type error
         self.graph.add_node(
-            name, instance=instance, input_sockets=input_sockets, output_sockets=output_sockets, visits=0
+            name,
+            instance=instance,
+            input_sockets=instance.__haystack_input__._sockets_dict,  # type: ignore[attr-defined]
+            output_sockets=instance.__haystack_output__._sockets_dict,  # type: ignore[attr-defined]
+            visits=0,
         )
 
     def connect(self, connect_from: str, connect_to: str) -> None:
@@ -381,6 +382,16 @@ class Pipeline:
         except KeyError as exc:
             raise ValueError(f"Component named {name} not found in the pipeline.") from exc
 
+    def get_component_name(self, instance: Component) -> str:
+        """
+        Returns the name of a Component instance. If the Component has not been added to this Pipeline,
+        returns an empty string.
+        """
+        for name, inst in self.graph.nodes(data="instance"):
+            if inst == instance:
+                return name
+        return ""
+
     def inputs(self) -> Dict[str, Dict[str, Any]]:
         """
         Returns a dictionary containing the inputs of a pipeline. Each key in the dictionary
@@ -465,16 +476,16 @@ class Pipeline:
             if component_name not in self.graph.nodes:
                 raise ValueError(f"Component named {component_name} not found in the pipeline.")
             instance = self.graph.nodes[component_name]["instance"]
-            for socket_name, socket in instance.__haystack_input__.items():
+            for socket_name, socket in instance.__haystack_input__._sockets_dict.items():
                 if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
                     raise ValueError(f"Missing input for component {component_name}: {socket_name}")
             for input_name in component_inputs.keys():
-                if input_name not in instance.__haystack_input__:
+                if input_name not in instance.__haystack_input__._sockets_dict:
                     raise ValueError(f"Input {input_name} not found in component {component_name}.")
 
         for component_name in self.graph.nodes:
             instance = self.graph.nodes[component_name]["instance"]
-            for socket_name, socket in instance.__haystack_input__.items():
+            for socket_name, socket in instance.__haystack_input__._sockets_dict.items():
                 component_inputs = data.get(component_name, {})
                 if socket.senders == [] and socket.is_mandatory and socket_name not in component_inputs:
                     raise ValueError(f"Missing input for component {component_name}: {socket_name}")
@@ -518,7 +529,7 @@ class Pipeline:
             for component_input, input_value in component_inputs.items():
                 # Handle mutable input data
                 data[component_name][component_input] = copy(input_value)
-                if instance.__haystack_input__[component_input].is_variadic:
+                if instance.__haystack_input__._sockets_dict[component_input].is_variadic:
                     # Components that have variadic inputs need to receive lists as input.
                     # We don't want to force the user to always pass lists, so we convert single values to lists here.
                     # If it's already a list we assume the component takes a variadic input of lists, so we
@@ -533,12 +544,12 @@ class Pipeline:
         for node_name in self.graph.nodes:
             component = self.graph.nodes[node_name]["instance"]
 
-            if len(component.__haystack_input__) == 0:
+            if len(component.__haystack_input__._sockets_dict) == 0:
                 # Component has no input, can run right away
                 to_run.append((node_name, component))
                 continue
 
-            for socket in component.__haystack_input__.values():
+            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))
@@ -561,12 +572,12 @@ class Pipeline:
         while len(to_run) > 0:
             name, comp = to_run.pop(0)
 
-            if any(socket.is_variadic for socket in comp.__haystack_input__.values()) and not getattr(  # type: ignore
+            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__.values()):  # type: ignore
+                    if not any(socket.is_variadic for socket in other_comp.__haystack_input__._sockets_dict.values()):  # type: ignore
                         there_are_non_variadics = True
                         break
 
@@ -575,7 +586,7 @@ class Pipeline:
                         waiting_for_input.append((name, comp))
                     continue
 
-            if name in last_inputs and len(comp.__haystack_input__) == len(last_inputs[name]):  # type: ignore
+            if name in last_inputs and len(comp.__haystack_input__._sockets_dict) == len(last_inputs[name]):  # type: ignore
                 # This component has all the inputs it needs to run
                 res = comp.run(**last_inputs[name])
 
@@ -649,7 +660,7 @@ class Pipeline:
                     # This is our last resort, if there's no lazy variadic 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__.values())  # type: ignore
+                        is_variadic = any(socket.is_variadic for socket in comp.__haystack_input__._sockets_dict.values())  # type: ignore
                         if is_variadic and not getattr(comp, "is_greedy", False):
                             break
                     else:
@@ -680,14 +691,14 @@ class Pipeline:
                         last_inputs[name] = {}
 
                     # Lazy variadics must be removed only if there's nothing else to run at this stage
-                    is_variadic = any(socket.is_variadic for socket in comp.__haystack_input__.values())  # type: ignore
+                    is_variadic = any(socket.is_variadic for socket in comp.__haystack_input__._sockets_dict.values())  # type: ignore
                     if is_variadic and not getattr(comp, "is_greedy", False):
                         there_are_only_lazy_variadics = True
                         for other_name, other_comp in waiting_for_input:
                             if name == other_name:
                                 continue
                             there_are_only_lazy_variadics &= any(
-                                socket.is_variadic for socket in other_comp.__haystack_input__.values()  # type: ignore
+                                socket.is_variadic for socket in other_comp.__haystack_input__._sockets_dict.values()  # type: ignore
                             ) and not getattr(other_comp, "is_greedy", False)
 
                         if not there_are_only_lazy_variadics:
@@ -695,7 +706,7 @@ class Pipeline:
 
                     # Find the first component that has all the inputs it needs to run
                     has_enough_inputs = True
-                    for input_socket in comp.__haystack_input__.values():  # type: ignore
+                    for input_socket in comp.__haystack_input__._sockets_dict.values():  # type: ignore
                         if input_socket.is_mandatory and input_socket.name not in last_inputs[name]:
                             has_enough_inputs = False
                             break
diff --git a/haystack/utils/hf.py b/haystack/utils/hf.py
index 29240762..32d9be1b 100644
--- a/haystack/utils/hf.py
+++ b/haystack/utils/hf.py
@@ -1,17 +1,29 @@
 import copy
+import inspect
 import logging
-from typing import Any, Dict, Optional
+from enum import Enum
+from typing import Any, Dict, Optional, List, Union, Callable
+
+from haystack.dataclasses import StreamingChunk
 from haystack.lazy_imports import LazyImport
 from haystack.utils.device import ComponentDevice
-
+from haystack.utils.auth import Secret
 
 with LazyImport(message="Run 'pip install transformers[torch]'") as torch_import:
     import torch
 
+with LazyImport(message="Run 'pip install transformers'") as transformers_import:
+    from huggingface_hub.utils import RepositoryNotFoundError
+    from huggingface_hub import InferenceClient, HfApi
 
 logger = logging.getLogger(__name__)
 
 
+class HFModelType(Enum):
+    EMBEDDING = 1
+    GENERATION = 2
+
+
 def serialize_hf_model_kwargs(kwargs: Dict[str, Any]):
     """
     Recursively serialize HuggingFace specific model keyword arguments
@@ -78,3 +90,129 @@ def resolve_hf_device_map(device: Optional[ComponentDevice], model_kwargs: Optio
     model_kwargs["device_map"] = device_map
 
     return model_kwargs
+
+
+def check_valid_model(model_id: str, model_type: HFModelType, token: Optional[Secret]) -> None:
+    """
+    Check if the provided model ID corresponds to a valid model on HuggingFace Hub.
+    Also check if the model is an embedding or generation model.
+
+    :param model_id: A string representing the HuggingFace model ID.
+    :param model_type: the model type, HFModelType.EMBEDDING or HFModelType.GENERATION
+    :param token: The optional authentication token.
+    :raises ValueError: If the model is not found or is not a embedding model.
+    """
+    transformers_import.check()
+
+    api = HfApi()
+    try:
+        model_info = api.model_info(model_id, token=token.resolve_value() if token else None)
+    except RepositoryNotFoundError as e:
+        raise ValueError(
+            f"Model {model_id} not found on HuggingFace Hub. Please provide a valid HuggingFace model_id."
+        ) from e
+
+    if model_type == HFModelType.EMBEDDING:
+        allowed_model = model_info.pipeline_tag in ["sentence-similarity", "feature-extraction"]
+        error_msg = f"Model {model_id} is not a embedding model. Please provide a embedding model."
+    elif model_type == HFModelType.GENERATION:
+        allowed_model = model_info.pipeline_tag in ["text-generation", "text2text-generation"]
+        error_msg = f"Model {model_id} is not a text generation model. Please provide a text generation model."
+
+    if not allowed_model:
+        raise ValueError(error_msg)
+
+
+def check_generation_params(kwargs: Optional[Dict[str, Any]], additional_accepted_params: Optional[List[str]] = None):
+    """
+    Check the provided generation parameters for validity.
+
+    :param kwargs: A dictionary containing the generation parameters.
+    :param additional_accepted_params: An optional list of strings representing additional accepted parameters.
+    :raises ValueError: If any unknown text generation parameters are provided.
+    """
+    transformers_import.check()
+
+    if kwargs:
+        accepted_params = {
+            param
+            for param in inspect.signature(InferenceClient.text_generation).parameters.keys()
+            if param not in ["self", "prompt"]
+        }
+        if additional_accepted_params:
+            accepted_params.update(additional_accepted_params)
+        unknown_params = set(kwargs.keys()) - accepted_params
+        if unknown_params:
+            raise ValueError(
+                f"Unknown text generation parameters: {unknown_params}. The valid parameters are: {accepted_params}."
+            )
+
+
+with LazyImport(message="Run 'pip install transformers[torch]'") as torch_and_transformers_import:
+    from transformers import StoppingCriteria, PreTrainedTokenizer, PreTrainedTokenizerFast, TextStreamer
+
+    transformers_import.check()
+    torch_import.check()
+
+    class StopWordsCriteria(StoppingCriteria):
+        """
+        Stops text generation if any one of the stop words is generated.
+
+        Note: When a stop word is encountered, the generation of new text is stopped.
+        However, if the stop word is in the prompt itself, it can stop generating new text
+        prematurely after the first token. This is particularly important for LLMs designed
+        for dialogue generation. For these models, like for example mosaicml/mpt-7b-chat,
+        the output includes both the new text and the original prompt. Therefore, it's important
+        to make sure your prompt has no stop words.
+        """
+
+        def __init__(
+            self,
+            tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
+            stop_words: List[str],
+            device: Union[str, torch.device] = "cpu",
+        ):
+            super().__init__()
+            # check if tokenizer is a valid tokenizer
+            if not isinstance(tokenizer, (PreTrainedTokenizer, PreTrainedTokenizerFast)):
+                raise ValueError(
+                    f"Invalid tokenizer provided for StopWordsCriteria - {tokenizer}. "
+                    f"Please provide a valid tokenizer from the HuggingFace Transformers library."
+                )
+            if not tokenizer.pad_token:
+                if tokenizer.eos_token:
+                    tokenizer.pad_token = tokenizer.eos_token
+                else:
+                    tokenizer.add_special_tokens({"pad_token": "[PAD]"})
+            encoded_stop_words = tokenizer(stop_words, add_special_tokens=False, padding=True, return_tensors="pt")
+            self.stop_ids = encoded_stop_words.input_ids.to(device)
+
+        def __call__(self, input_ids: torch.LongTensor, scores: torch.FloatTensor, **kwargs) -> bool:
+            for stop_id in self.stop_ids:
+                found_stop_word = self.is_stop_word_found(input_ids, stop_id)
+                if found_stop_word:
+                    return True
+            return False
+
+        def is_stop_word_found(self, generated_text_ids: torch.Tensor, stop_id: torch.Tensor) -> bool:
+            generated_text_ids = generated_text_ids[-1]
+            len_generated_text_ids = generated_text_ids.size(0)
+            len_stop_id = stop_id.size(0)
+            result = all(generated_text_ids[len_generated_text_ids - len_stop_id :].eq(stop_id))
+            return result
+
+    class HFTokenStreamingHandler(TextStreamer):
+        def __init__(
+            self,
+            tokenizer: Union[PreTrainedTokenizer, PreTrainedTokenizerFast],
+            stream_handler: Callable[[StreamingChunk], None],
+            stop_words: Optional[List[str]] = None,
+        ):
+            super().__init__(tokenizer=tokenizer, skip_prompt=True)  # type: ignore
+            self.token_handler = stream_handler
+            self.stop_words = stop_words or []
+
+        def on_finalized_text(self, word: str, stream_end: bool = False):
+            word_to_send = word + "\n" if stream_end else word
+            if word_to_send.strip() not in self.stop_words:
+                self.token_handler(StreamingChunk(content=word_to_send))
diff --git a/releasenotes/notes/merge-hf-utils-modules-5c16e04025123568.yaml b/releasenotes/notes/merge-hf-utils-modules-5c16e04025123568.yaml
new file mode 100644
index 00000000..d57cdfcd
--- /dev/null
+++ b/releasenotes/notes/merge-hf-utils-modules-5c16e04025123568.yaml
@@ -0,0 +1,5 @@
+---
+enhancements:
+  - |
+    Code from different "hf_utils.py" modules spread across different packages was
+    merged into `haystack.utils.hf`.


Test patch:
diff --git a/haystack/testing/sample_components/repeat.py b/haystack/testing/sample_components/repeat.py
index 73e25097..a1f62879 100644
--- a/haystack/testing/sample_components/repeat.py
+++ b/haystack/testing/sample_components/repeat.py
@@ -9,11 +9,11 @@ from haystack.core.component import component
 @component
 class Repeat:
     def __init__(self, outputs: List[str]):
-        self.outputs = outputs
+        self._outputs = outputs
         component.set_output_types(self, **{k: int for k in outputs})
 
     def run(self, value: int):
         """
         :param value: the value to repeat.
         """
-        return {val: value for val in self.outputs}
+        return {val: value for val in self._outputs}
diff --git a/test/components/builders/test_dynamic_chat_prompt_builder.py b/test/components/builders/test_dynamic_chat_prompt_builder.py
index 67183d02..d75e0051 100644
--- a/test/components/builders/test_dynamic_chat_prompt_builder.py
+++ b/test/components/builders/test_dynamic_chat_prompt_builder.py
@@ -15,16 +15,16 @@ class TestDynamicChatPromptBuilder:
 
         # we have inputs that contain: prompt_source, template_variables + runtime_variables
         expected_keys = set(runtime_variables + ["prompt_source", "template_variables"])
-        assert set(builder.__haystack_input__.keys()) == expected_keys
+        assert set(builder.__haystack_input__._sockets_dict.keys()) == expected_keys
 
         # response is always prompt regardless of chat mode
-        assert set(builder.__haystack_output__.keys()) == {"prompt"}
+        assert set(builder.__haystack_output__._sockets_dict.keys()) == {"prompt"}
 
         # prompt_source is a list of ChatMessage
-        assert builder.__haystack_input__["prompt_source"].type == List[ChatMessage]
+        assert builder.__haystack_input__._sockets_dict["prompt_source"].type == List[ChatMessage]
 
         # output is always prompt, but the type is different depending on the chat mode
-        assert builder.__haystack_output__["prompt"].type == List[ChatMessage]
+        assert builder.__haystack_output__._sockets_dict["prompt"].type == List[ChatMessage]
 
     def test_non_empty_chat_messages(self):
         prompt_builder = DynamicChatPromptBuilder(runtime_variables=["documents"])
diff --git a/test/components/builders/test_dynamic_prompt_builder.py b/test/components/builders/test_dynamic_prompt_builder.py
index c3508b25..7afacd2c 100644
--- a/test/components/builders/test_dynamic_prompt_builder.py
+++ b/test/components/builders/test_dynamic_prompt_builder.py
@@ -16,16 +16,16 @@ class TestDynamicPromptBuilder:
         # regardless of the chat mode
         # we have inputs that contain: prompt_source, template_variables + runtime_variables
         expected_keys = set(runtime_variables + ["prompt_source", "template_variables"])
-        assert set(builder.__haystack_input__.keys()) == expected_keys
+        assert set(builder.__haystack_input__._sockets_dict.keys()) == expected_keys
 
         # response is always prompt regardless of chat mode
-        assert set(builder.__haystack_output__.keys()) == {"prompt"}
+        assert set(builder.__haystack_output__._sockets_dict.keys()) == {"prompt"}
 
         # prompt_source is a list of ChatMessage or a string
-        assert builder.__haystack_input__["prompt_source"].type == str
+        assert builder.__haystack_input__._sockets_dict["prompt_source"].type == str
 
         # output is always prompt, but the type is different depending on the chat mode
-        assert builder.__haystack_output__["prompt"].type == str
+        assert builder.__haystack_output__._sockets_dict["prompt"].type == str
 
     def test_processing_a_simple_template_with_provided_variables(self):
         runtime_variables = ["var1", "var2", "var3"]
diff --git a/test/components/generators/test_hf_utils.py b/test/components/generators/test_hf_utils.py
index c50c2af6..01a8eeba 100644
--- a/test/components/generators/test_hf_utils.py
+++ b/test/components/generators/test_hf_utils.py
@@ -1,6 +1,6 @@
 import pytest
 
-from haystack.components.generators.hf_utils import check_generation_params
+from haystack.utils.hf import check_generation_params
 
 
 def test_empty_dictionary():
diff --git a/test/components/routers/test_conditional_router.py b/test/components/routers/test_conditional_router.py
index 726d7cd4..dcb52cff 100644
--- a/test/components/routers/test_conditional_router.py
+++ b/test/components/routers/test_conditional_router.py
@@ -86,8 +86,8 @@ class TestRouter:
         router = ConditionalRouter(routes)
 
         assert router.routes == routes
-        assert set(router.__haystack_input__.keys()) == {"query", "streams"}
-        assert set(router.__haystack_output__.keys()) == {"query", "streams"}
+        assert set(router.__haystack_input__._sockets_dict.keys()) == {"query", "streams"}
+        assert set(router.__haystack_output__._sockets_dict.keys()) == {"query", "streams"}
 
     def test_router_evaluate_condition_expressions(self, router):
         # first route should be selected
diff --git a/test/core/component/test_component.py b/test/core/component/test_component.py
index a60ea6ba..bbe2605f 100644
--- a/test/core/component/test_component.py
+++ b/test/core/component/test_component.py
@@ -1,5 +1,4 @@
-import typing
-from typing import Any, Optional
+from typing import Any
 
 import pytest
 
@@ -89,6 +88,7 @@ def test_missing_run():
 
 
 def test_set_input_types():
+    @component
     class MockComponent:
         def __init__(self):
             component.set_input_types(self, value=Any)
@@ -105,7 +105,7 @@ def test_set_input_types():
             return {"value": 1}
 
     comp = MockComponent()
-    assert comp.__haystack_input__ == {"value": InputSocket("value", Any)}
+    assert comp.__haystack_input__._sockets_dict == {"value": InputSocket("value", Any)}
     assert comp.run() == {"value": 1}
 
 
@@ -126,7 +126,7 @@ def test_set_output_types():
             return {"value": 1}
 
     comp = MockComponent()
-    assert comp.__haystack_output__ == {"value": OutputSocket("value", int)}
+    assert comp.__haystack_output__._sockets_dict == {"value": OutputSocket("value", int)}
 
 
 def test_output_types_decorator_with_compatible_type():
@@ -144,7 +144,7 @@ def test_output_types_decorator_with_compatible_type():
             return cls()
 
     comp = MockComponent()
-    assert comp.__haystack_output__ == {"value": OutputSocket("value", int)}
+    assert comp.__haystack_output__._sockets_dict == {"value": OutputSocket("value", int)}
 
 
 def test_component_decorator_set_it_as_component():
@@ -173,8 +173,8 @@ def test_input_has_default_value():
             return {"value": value}
 
     comp = MockComponent()
-    assert comp.__haystack_input__["value"].default_value == 42
-    assert not comp.__haystack_input__["value"].is_mandatory
+    assert comp.__haystack_input__._sockets_dict["value"].default_value == 42
+    assert not comp.__haystack_input__._sockets_dict["value"].is_mandatory
 
 
 def test_keyword_only_args():
@@ -187,5 +187,5 @@ def test_keyword_only_args():
             return {"value": arg}
 
     comp = MockComponent()
-    component_inputs = {name: {"type": socket.type} for name, socket in comp.__haystack_input__.items()}
+    component_inputs = {name: {"type": socket.type} for name, socket in comp.__haystack_input__._sockets_dict.items()}
     assert component_inputs == {"arg": {"type": int}}
diff --git a/test/core/component/test_sockets.py b/test/core/component/test_sockets.py
new file mode 100644
index 00000000..ac3b01bd
--- /dev/null
+++ b/test/core/component/test_sockets.py
@@ -0,0 +1,57 @@
+import pytest
+
+from haystack.core.component.sockets import InputSocket, Sockets
+from haystack.core.pipeline import Pipeline
+from haystack.testing.factory import component_class
+
+
+class TestSockets:
+    def test_init(self):
+        comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
+        sockets = {"input_1": InputSocket("input_1", int), "input_2": InputSocket("input_2", int)}
+        io = Sockets(component=comp, sockets_dict=sockets, sockets_io_type=InputSocket)
+        assert io._component == comp
+        assert "input_1" in io.__dict__
+        assert io.__dict__["input_1"] == comp.__haystack_input__._sockets_dict["input_1"]
+        assert "input_2" in io.__dict__
+        assert io.__dict__["input_2"] == comp.__haystack_input__._sockets_dict["input_2"]
+
+    def test_init_with_empty_sockets(self):
+        comp = component_class("SomeComponent")()
+        io = Sockets(component=comp, sockets_dict={}, sockets_io_type=InputSocket)
+
+        assert io._component == comp
+        assert io._sockets_dict == {}
+
+    def test_component_name(self):
+        comp = component_class("SomeComponent")()
+        io = Sockets(component=comp, sockets_dict={}, sockets_io_type=InputSocket)
+        assert io._component_name() == str(comp)
+
+    def test_component_name_added_to_pipeline(self):
+        comp = component_class("SomeComponent")()
+        pipeline = Pipeline()
+        pipeline.add_component("my_component", comp)
+
+        io = Sockets(component=comp, sockets_dict={}, sockets_io_type=InputSocket)
+        assert io._component_name() == "my_component"
+
+    def test_getattribute(self):
+        comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
+        io = Sockets(component=comp, sockets_dict=comp.__haystack_input__._sockets_dict, sockets_io_type=InputSocket)
+
+        assert io.input_1 == comp.__haystack_input__._sockets_dict["input_1"]
+        assert io.input_2 == comp.__haystack_input__._sockets_dict["input_2"]
+
+    def test_getattribute_non_existing_socket(self):
+        comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
+        io = Sockets(component=comp, sockets_dict=comp.__haystack_input__._sockets_dict, sockets_io_type=InputSocket)
+
+        with pytest.raises(AttributeError):
+            io.input_3
+
+    def test_repr(self):
+        comp = component_class("SomeComponent", input_types={"input_1": int, "input_2": int})()
+        io = Sockets(component=comp, sockets_dict=comp.__haystack_input__._sockets_dict, sockets_io_type=InputSocket)
+        res = repr(io)
+        assert res == f"{comp} inputs:\n  - input_1: int\n  - input_2: int"
diff --git a/test/core/pipeline/test_connect.py b/test/core/pipeline/test_connect.py
new file mode 100644
index 00000000..92f0afc4
--- /dev/null
+++ b/test/core/pipeline/test_connect.py
@@ -0,0 +1,200 @@
+import pytest
+
+from haystack.core.component.types import Variadic
+from haystack.core.errors import PipelineConnectError
+from haystack.core.pipeline import Pipeline
+from haystack.core.pipeline.pipeline import parse_connect_string
+from haystack.testing import factory
+
+
+def test_parse_connection():
+    assert parse_connect_string("foobar") == ("foobar", None)
+    assert parse_connect_string("foo.bar") == ("foo", "bar")
+
+
+def test_connect():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1.value", "comp2.value")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_reconnection():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1.value", "comp2.value")
+    pipe.connect("comp1.value", "comp2.value")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_with_sender_component_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1", "comp2.value")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_with_receiver_component_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1.value", "comp2")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_with_sender_and_receiver_component_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1", "comp2")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_with_sender_not_in_pipeline():
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(ValueError):
+        pipe.connect("comp1.value", "comp2.value")
+
+
+def test_with_receiver_not_in_pipeline():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    with pytest.raises(ValueError):
+        pipe.connect("comp1.value", "comp2.value")
+
+
+def test_with_sender_socket_name_not_in_pipeline():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp1.non_existing", "comp2.value")
+
+
+def test_with_receiver_socket_name_not_in_pipeline():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp1.value", "comp2.non_existing")
+
+
+def test_with_no_matching_types_and_same_names():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": str})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp1", "comp2")
+
+
+def test_with_multiple_sender_connections_with_same_type_and_differing_name():
+    comp1 = factory.component_class("Comp1", output_types={"val1": int, "val2": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp1", "comp2")
+
+
+def test_with_multiple_receiver_connections_with_same_type_and_differing_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"val1": int, "val2": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp1", "comp2")
+
+
+def test_with_multiple_sender_connections_with_same_type_and_same_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int, "other": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1", "comp2")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_with_multiple_receiver_connections_with_same_type_and_same_name():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", input_types={"value": int, "other": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.connect("comp1", "comp2")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp2"]
+    assert comp2.__haystack_input__.value.senders == ["comp1"]
+    assert list(pipe.graph.edges) == [("comp1", "comp2", "value/value")]
+
+
+def test_multiple_outputs_to_non_variadic_input():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", output_types={"value": int})()
+    comp3 = factory.component_class("Comp3", input_types={"value": int})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.add_component("comp3", comp3)
+    pipe.connect("comp1.value", "comp3.value")
+    with pytest.raises(PipelineConnectError):
+        pipe.connect("comp2.value", "comp3.value")
+
+
+def test_multiple_outputs_to_variadic_input():
+    comp1 = factory.component_class("Comp1", output_types={"value": int})()
+    comp2 = factory.component_class("Comp2", output_types={"value": int})()
+    comp3 = factory.component_class("Comp3", input_types={"value": Variadic[int]})()
+    pipe = Pipeline()
+    pipe.add_component("comp1", comp1)
+    pipe.add_component("comp2", comp2)
+    pipe.add_component("comp3", comp3)
+    pipe.connect("comp1.value", "comp3.value")
+    pipe.connect("comp2.value", "comp3.value")
+
+    assert comp1.__haystack_output__.value.receivers == ["comp3"]
+    assert comp2.__haystack_output__.value.receivers == ["comp3"]
+    assert comp3.__haystack_input__.value.senders == ["comp1", "comp2"]
+    assert list(pipe.graph.edges) == [("comp1", "comp3", "value/value"), ("comp2", "comp3", "value/value")]
diff --git a/test/core/pipeline/test_connections.py b/test/core/pipeline/test_connections.py
deleted file mode 100644
index 8a795629..00000000
--- a/test/core/pipeline/test_connections.py
+++ /dev/null
@@ -1,404 +0,0 @@
-# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
-#
-# SPDX-License-Identifier: Apache-2.0
-import re
-from enum import Enum
-from pathlib import Path
-from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union
-
-import pytest
-
-from haystack.core.errors import PipelineConnectError
-from haystack.core.pipeline import Pipeline
-from haystack.core.pipeline.pipeline import parse_connect_string
-from haystack.testing import factory
-from haystack.testing.sample_components import AddFixedValue
-
-
-class Class1:
-    ...
-
-
-class Class2:
-    ...
-
-
-class Class3(Class1):
-    ...
-
-
-class Enum1(Enum):
-    TEST1 = Class1
-    TEST2 = Class2
-
-
-@pytest.mark.parametrize(
-    "from_type,to_type",
-    [
-        pytest.param(str, str, id="same-primitives"),
-        pytest.param(str, Optional[str], id="receiving-primitive-is-optional"),
-        pytest.param(str, Union[int, str], id="receiving-type-is-union-of-primitives"),
-        pytest.param(Union[int, str], Union[int, str], id="identical-unions"),
-        pytest.param(Union[int, str], Union[int, str, bool], id="receiving-union-is-superset-of-sender"),
-        pytest.param(str, Any, id="primitive-to-any"),
-        pytest.param(Class1, Class1, id="same-class"),
-        pytest.param(Class1, Optional[Class1], id="receiving-class-is-optional"),
-        pytest.param(Class1, Class1, id="class-to-any"),
-        pytest.param(Class3, Class1, id="subclass-to-class"),
-        pytest.param(Class1, Union[int, Class1], id="receiving-type-is-union-of-classes"),
-        pytest.param(Class3, Union[int, Class1], id="receiving-type-is-union-of-superclasses"),
-        pytest.param(List[int], List[int], id="same-lists"),
-        pytest.param(List[int], Optional[List[int]], id="receiving-list-is-optional"),
-        pytest.param(List[int], List[Any], id="list-of-primitive-to-list-of-any"),
-        pytest.param(List[Class1], List[Class1], id="list-of-same-classes"),
-        pytest.param(List[Class3], List[Class1], id="list-of-subclass-to-list-of-class"),
-        pytest.param(List[Class1], List[Any], id="list-of-classes-to-list-of-any"),
-        pytest.param(List[Set[Sequence[bool]]], List[Set[Sequence[bool]]], id="nested-sequences-of-same-primitives"),
-        pytest.param(
-            List[Set[Sequence[bool]]],
-            List[Set[Sequence[Any]]],
-            id="nested-sequences-of-primitives-to-nested-sequences-of-any",
-        ),
-        pytest.param(List[Set[Sequence[Class1]]], List[Set[Sequence[Class1]]], id="nested-sequences-of-same-classes"),
-        pytest.param(
-            List[Set[Sequence[Class3]]],
-            List[Set[Sequence[Class1]]],
-            id="nested-sequences-of-subclasses-to-nested-sequences-of-classes",
-        ),
-        pytest.param(
-            List[Set[Sequence[Class1]]],
-            List[Set[Sequence[Any]]],
-            id="nested-sequences-of-classes-to-nested-sequences-of-any",
-        ),
-        pytest.param(Dict[str, int], Dict[str, int], id="same-dicts-of-primitives"),
-        pytest.param(Dict[str, int], Dict[Any, int], id="dict-of-primitives-to-dict-of-any-keys"),
-        pytest.param(Dict[str, int], Dict[str, Any], id="dict-of-primitives-to-dict-of-any-values"),
-        pytest.param(Dict[str, int], Dict[Any, Any], id="dict-of-primitives-to-dict-of-any-key-and-values"),
-        pytest.param(Dict[str, Class1], Dict[str, Class1], id="same-dicts-of-classes-values"),
-        pytest.param(Dict[str, Class3], Dict[str, Class1], id="dict-of-subclasses-to-dict-of-classes"),
-        pytest.param(Dict[str, Class1], Dict[Any, Class1], id="dict-of-classes-to-dict-of-any-keys"),
-        pytest.param(Dict[str, Class1], Dict[str, Any], id="dict-of-classes-to-dict-of-any-values"),
-        pytest.param(Dict[str, Class1], Dict[Any, Any], id="dict-of-classes-to-dict-of-any-key-and-values"),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[str, Dict[str, int]]],
-            id="nested-mappings-of-same-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[str, Dict[Any, int]]],
-            id="nested-mapping-of-primitives-to-nested-mapping-of-any-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[Any, Dict[str, int]]],
-            id="nested-mapping-of-primitives-to-nested-mapping-of-higher-level-any-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[str, Dict[str, Any]]],
-            id="nested-mapping-of-primitives-to-nested-mapping-of-any-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[Any, Dict[Any, Any]]],
-            id="nested-mapping-of-primitives-to-nested-mapping-of-any-keys-and-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            id="nested-mappings-of-same-classes",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class3]]],
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            id="nested-mapping-of-subclasses-to-nested-mapping-of-classes",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[str, Dict[Any, Class1]]],
-            id="nested-mapping-of-classes-to-nested-mapping-of-any-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[Any, Dict[str, Class1]]],
-            id="nested-mapping-of-classes-to-nested-mapping-of-higher-level-any-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[str, Dict[str, Any]]],
-            id="nested-mapping-of-classes-to-nested-mapping-of-any-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[Any, Dict[Any, Any]]],
-            id="nested-mapping-of-classes-to-nested-mapping-of-any-keys-and-values",
-        ),
-        pytest.param(Literal["a", "b", "c"], Literal["a", "b", "c"], id="same-primitive-literal"),
-        pytest.param(Literal[Enum1.TEST1], Literal[Enum1.TEST1], id="same-enum-literal"),
-        pytest.param(
-            Tuple[Optional[Literal["a", "b", "c"]], Union[Path, Dict[int, Class1]]],
-            Tuple[Optional[Literal["a", "b", "c"]], Union[Path, Dict[int, Class1]]],
-            id="identical-deeply-nested-complex-type",
-        ),
-    ],
-)
-def test_connect_compatible_types(from_type, to_type):
-    comp1 = factory.component_class("Comp1", output_types={"value": from_type})()
-    comp2 = factory.component_class("Comp2", input_types={"value": to_type})()
-
-    pipe = Pipeline()
-    pipe.add_component("c1", comp1)
-    pipe.add_component("c2", comp2)
-    pipe.connect("c1", "c2")
-    assert list(pipe.graph.edges) == [("c1", "c2", "value/value")]
-
-
-@pytest.mark.parametrize(
-    "from_type, to_type",
-    [
-        pytest.param(int, bool, id="different-primitives"),
-        pytest.param(Class1, Class2, id="different-classes"),
-        pytest.param(Class1, Class3, id="class-to-subclass"),
-        pytest.param(Any, int, id="any-to-primitive"),
-        pytest.param(Any, Class2, id="any-to-class"),
-        pytest.param(Optional[str], str, id="sending-primitive-is-optional"),
-        pytest.param(Optional[Class1], Class1, id="sending-class-is-optional"),
-        pytest.param(Optional[List[int]], List[int], id="sending-list-is-optional"),
-        pytest.param(Union[int, str], str, id="sending-type-is-union"),
-        pytest.param(Union[int, str, bool], Union[int, str], id="sending-union-is-superset-of-receiver"),
-        pytest.param(Union[int, bool], Union[int, str], id="partially-overlapping-unions-with-primitives"),
-        pytest.param(Union[int, Class1], Union[int, Class2], id="partially-overlapping-unions-with-classes"),
-        pytest.param(List[int], List[str], id="different-lists-of-primitives"),
-        pytest.param(List[int], List, id="list-of-primitive-to-bare-list"),  # is "correct", but we don't support it
-        pytest.param(List[int], list, id="list-of-primitive-to-list-object"),  # is "correct", but we don't support it
-        pytest.param(List[Class1], List[Class2], id="different-lists-of-classes"),
-        pytest.param(List[Class1], List[Class3], id="lists-of-classes-to-subclasses"),
-        pytest.param(List[Any], List[str], id="list-of-any-to-list-of-primitives"),
-        pytest.param(List[Any], List[Class2], id="list-of-any-to-list-of-classes"),
-        pytest.param(
-            List[Set[Sequence[str]]], List[Set[Sequence[bool]]], id="nested-sequences-of-different-primitives"
-        ),
-        pytest.param(
-            List[Set[Sequence[str]]], Set[List[Sequence[str]]], id="different-nested-sequences-of-same-primitives"
-        ),
-        pytest.param(
-            List[Set[Sequence[Class1]]], List[Set[Sequence[Class2]]], id="nested-sequences-of-different-classes"
-        ),
-        pytest.param(
-            List[Set[Sequence[Class1]]], List[Set[Sequence[Class3]]], id="nested-sequences-of-classes-to-subclasses"
-        ),
-        pytest.param(
-            List[Set[Sequence[Class1]]], Set[List[Sequence[Class1]]], id="different-nested-sequences-of-same-class"
-        ),
-        pytest.param(
-            List[Set[Sequence[Any]]], List[Set[Sequence[bool]]], id="nested-list-of-Any-to-nested-list-of-primitives"
-        ),
-        pytest.param(
-            List[Set[Sequence[Any]]], List[Set[Sequence[Class2]]], id="nested-list-of-Any-to-nested-list-of-classes"
-        ),
-        pytest.param(Dict[str, int], Dict[int, int], id="different-dict-of-primitive-keys"),
-        pytest.param(Dict[str, int], Dict[str, bool], id="different-dict-of-primitive-values"),
-        pytest.param(Dict[str, Class1], Dict[str, Class2], id="different-dict-of-class-values"),
-        pytest.param(Dict[str, Class1], Dict[str, Class3], id="different-dict-of-class-to-subclass-values"),
-        pytest.param(Dict[Any, int], Dict[int, int], id="dict-of-Any-keys-to-dict-of-primitives"),
-        pytest.param(Dict[str, Any], Dict[int, int], id="dict-of-Any-values-to-dict-of-primitives"),
-        pytest.param(Dict[str, Any], Dict[int, Class1], id="dict-of-Any-values-to-dict-of-classes"),
-        pytest.param(Dict[Any, Any], Dict[int, int], id="dict-of-Any-keys-and-values-to-dict-of-primitives"),
-        pytest.param(Dict[Any, Any], Dict[int, Class1], id="dict-of-Any-keys-and-values-to-dict-of-classes"),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Mapping[str, Dict[str, Dict[str, int]]],
-            id="different-nested-mappings-of-same-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[str, Dict[int, int]]],
-            id="same-nested-mappings-of-different-primitive-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[int, Dict[str, int]]],
-            id="same-nested-mappings-of-different-higer-level-primitive-keys",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            Dict[str, Mapping[str, Dict[str, bool]]],
-            id="same-nested-mappings-of-different-primitive-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[str, Dict[str, Class2]]],
-            id="same-nested-mappings-of-different-class-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            Dict[str, Mapping[str, Dict[str, Class2]]],
-            id="same-nested-mappings-of-class-to-subclass-values",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[Any, int]]],
-            Dict[str, Mapping[str, Dict[str, int]]],
-            id="nested-mapping-of-Any-keys-to-nested-mapping-of-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[Any, Dict[Any, int]]],
-            Dict[str, Mapping[str, Dict[str, int]]],
-            id="nested-mapping-of-higher-level-Any-keys-to-nested-mapping-of-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Any]]],
-            Dict[str, Mapping[str, Dict[str, int]]],
-            id="nested-mapping-of-Any-values-to-nested-mapping-of-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Any]]],
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            id="nested-mapping-of-Any-values-to-nested-mapping-of-classes",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[Any, Any]]],
-            Dict[str, Mapping[str, Dict[str, int]]],
-            id="nested-mapping-of-Any-keys-and-values-to-nested-mapping-of-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[Any, Any]]],
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            id="nested-mapping-of-Any-keys-and-values-to-nested-mapping-of-classes",
-        ),
-        pytest.param(Literal["a", "b", "c"], Literal["x", "y"], id="different-literal-of-same-primitive"),
-        pytest.param(Literal["a", "b", "c"], Literal["a", "b"], id="subset-literal"),
-        pytest.param(Literal[Enum1.TEST1], Literal[Enum1.TEST2], id="different-literal-of-same-enum"),
-        pytest.param(
-            Tuple[Optional[Literal["a", "b", "c"]], Union[Path, Dict[int, Class1]]],
-            Tuple[Literal["a", "b", "c"], Union[Path, Dict[int, Class1]]],
-            id="deeply-nested-complex-type-is-compatible-but-cannot-be-checked",
-        ),
-    ],
-)
-def test_connect_non_compatible_types(from_type, to_type):
-    comp1 = factory.component_class("Comp1", output_types={"value": from_type})()
-    comp2 = factory.component_class("Comp2", input_types={"value": to_type})()
-
-    pipe = Pipeline()
-    pipe.add_component("c1", comp1)
-    pipe.add_component("c2", comp2)
-
-    with pytest.raises(
-        PipelineConnectError,
-        match="Cannot connect 'c1.value' with 'c2.value': their declared input and output types do not match.",
-    ):
-        pipe.connect("c1", "c2")
-
-
-def test_connect_sender_component_does_not_exist():
-    add_1 = AddFixedValue()
-    add_2 = AddFixedValue()
-
-    pipe = Pipeline()
-    pipe.add_component("first", add_1)
-    pipe.add_component("second", add_2)
-    with pytest.raises(ValueError, match="Component named third not found in the pipeline"):
-        pipe.connect("third", "second")
-
-
-def test_connect_receiver_component_does_not_exist():
-    add_1 = AddFixedValue()
-    add_2 = AddFixedValue()
-
-    pipe = Pipeline()
-    pipe.add_component("first", add_1)
-    pipe.add_component("second", add_2)
-    with pytest.raises(ValueError, match="Component named third not found in the pipeline"):
-        pipe.connect("first", "third")
-
-
-def test_connect_sender_socket_does_not_exist():
-    add_1 = AddFixedValue()
-    add_2 = AddFixedValue()
-
-    pipe = Pipeline()
-    pipe.add_component("first", add_1)
-    pipe.add_component("second", add_2)
-    with pytest.raises(PipelineConnectError, match="first.wrong does not exist"):
-        pipe.connect("first.wrong", "second")
-
-
-def test_connect_receiver_socket_does_not_exist():
-    add_1 = AddFixedValue()
-    add_2 = AddFixedValue()
-
-    pipe = Pipeline()
-    pipe.add_component("first", add_1)
-    pipe.add_component("second", add_2)
-    with pytest.raises(PipelineConnectError, match="second.wrong does not exist"):
-        pipe.connect("first", "second.wrong")
-
-
-def test_connect_many_outputs_to_the_same_input():
-    add_1 = AddFixedValue()
-    add_2 = AddFixedValue()
-    add_3 = AddFixedValue()
-
-    pipe = Pipeline()
-    pipe.add_component("first", add_1)
-    pipe.add_component("second", add_2)
-    pipe.add_component("third", add_3)
-    pipe.connect("first.result", "second.value")
-    with pytest.raises(PipelineConnectError, match=r"second.value is already connected to \['first'\]"):
-        pipe.connect("third.result", "second.value")
-
-
-def test_connect_many_connections_possible_name_matches():
-    Component1 = factory.component_class("Component1", output_types={"value": str})
-    Component2 = factory.component_class(
-        "Component2", input_types={"value": str, "othervalue": str, "yetanothervalue": str}
-    )
-
-    pipe = Pipeline()
-    pipe.add_component("c1", Component1())
-    pipe.add_component("c2", Component2())
-    pipe.connect("c1", "c2")
-    assert list(pipe.graph.edges) == [("c1", "c2", "value/value")]
-
-
-def test_connect_many_connections_possible_no_name_matches():
-    Component1 = factory.component_class("Component1", output_types={"value": str})
-    Component2 = factory.component_class("Component2", input_types={"value1": str, "value2": str, "value3": str})
-
-    expected_message = re.escape(
-        """Cannot connect 'c1' with 'c2': more than one connection is possible between these components. Please specify the connection name, like: pipeline.connect('c1.value', 'c2.value1').
-'c1':
- - value: str
-'c2':
- - value1: str (available)
- - value2: str (available)
- - value3: str (available)"""
-    )
-
-    pipe = Pipeline()
-    pipe.add_component("c1", Component1())
-    pipe.add_component("c2", Component2())
-    with pytest.raises(PipelineConnectError, match=expected_message):
-        pipe.connect("c1", "c2")
-
-
-def test_parse_connection():
-    assert parse_connect_string("foobar") == ("foobar", None)
-    assert parse_connect_string("foo.bar") == ("foo", "bar")
-    assert parse_connect_string("foo.bar.baz") == ("foo", "bar.baz")
-
-
-def test_connect_with_same_socket_names():
-    SimpleComponent = factory.component_class("SimpleComponent", output_types={"documents": List})
-    ComponentWithMultipleInputs = factory.component_class(
-        "ComponentWithMultipleInputs", input_types={"question": Any, "documents": Any}
-    )
-
-    pipe = Pipeline()
-    pipe.add_component("simple", SimpleComponent())
-    pipe.add_component("multiple", ComponentWithMultipleInputs())
-
-    pipe.connect("simple", "multiple")
-
-    assert list(pipe.graph.edges) == [("simple", "multiple", "documents/documents")]
diff --git a/test/core/pipeline/test_pipeline.py b/test/core/pipeline/test_pipeline.py
index 5930f5e5..45f1c883 100644
--- a/test/core/pipeline/test_pipeline.py
+++ b/test/core/pipeline/test_pipeline.py
@@ -6,7 +6,7 @@ from typing import Optional
 
 import pytest
 
-from haystack.core.component.sockets import InputSocket, OutputSocket
+from haystack.core.component.types import InputSocket, OutputSocket
 from haystack.core.errors import PipelineError, PipelineRuntimeError
 from haystack.core.pipeline import Pipeline
 from haystack.testing.factory import component_class
@@ -28,6 +28,21 @@ def test_add_component_to_different_pipelines():
         second_pipe.add_component("some", some_component)
 
 
+def test_get_component_name():
+    pipe = Pipeline()
+    some_component = component_class("Some")()
+    pipe.add_component("some", some_component)
+
+    assert pipe.get_component_name(some_component) == "some"
+
+
+def test_get_component_name_not_added_to_pipeline():
+    pipe = Pipeline()
+    some_component = component_class("Some")()
+
+    assert pipe.get_component_name(some_component) == ""
+
+
 def test_run_with_component_that_does_not_return_dict():
     BrokenComponent = component_class(
         "BrokenComponent", input_types={"a": int}, output_types={"b": int}, output=1  # type:ignore
diff --git a/test/core/pipeline/test_reconnect.py b/test/core/pipeline/test_reconnect.py
deleted file mode 100644
index 80e73b0e..00000000
--- a/test/core/pipeline/test_reconnect.py
+++ /dev/null
@@ -1,21 +0,0 @@
-import pytest
-
-from haystack import Pipeline
-from haystack.core.errors import PipelineConnectError
-from haystack.testing.sample_components import Double
-
-
-def test_connect_component_twice():
-    pipe = Pipeline()
-    c1 = Double()
-    c2 = Double()
-    c3 = Double()
-    pipe.add_component("c1", c1)
-    pipe.add_component("c2", c2)
-    pipe.add_component("c3", c3)
-    pipe.connect("c1.value", "c2.value")
-    # the following should be a no-op
-    pipe.connect("c1.value", "c2.value")
-    # this should fail instead
-    with pytest.raises(PipelineConnectError):
-        pipe.connect("c3.value", "c2.value")
diff --git a/test/core/pipeline/test_type_utils.py b/test/core/pipeline/test_type_utils.py
new file mode 100644
index 00000000..473c5bad
--- /dev/null
+++ b/test/core/pipeline/test_type_utils.py
@@ -0,0 +1,326 @@
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Literal, Mapping, Optional, Sequence, Set, Tuple, Union
+
+import pytest
+
+from haystack.core.type_utils import _type_name, _types_are_compatible
+
+
+class Class1:
+    pass
+
+
+class Class2:
+    pass
+
+
+class Class3(Class1):
+    pass
+
+
+class Enum1(Enum):
+    TEST1 = Class1
+    TEST2 = Class2
+
+
+@pytest.mark.parametrize(
+    "type_,repr",
+    [
+        pytest.param(str, "str", id="primitive-types"),
+        pytest.param(Any, "Any", id="any"),
+        pytest.param(Class1, "Class1", id="class"),
+        pytest.param((Optional[int]), "Optional[int]", id="shallow-optional-with-primitive"),
+        pytest.param((Optional[Any]), "Optional[Any]", id="shallow-optional-with-any"),
+        pytest.param((Optional[Class1]), "Optional[Class1]", id="shallow-optional-with-class"),
+        pytest.param((Union[(bool, Class1)]), "Union[bool, Class1]", id="shallow-union"),
+        pytest.param((List[str]), "List[str]", id="shallow-sequence-of-primitives"),
+        pytest.param((List[Set[Sequence[str]]]), "List[Set[Sequence[str]]]", id="nested-sequence-of-primitives"),
+        pytest.param(
+            (Optional[List[Set[Sequence[str]]]]),
+            "Optional[List[Set[Sequence[str]]]]",
+            id="optional-nested-sequence-of-primitives",
+        ),
+        pytest.param(
+            (List[Set[Sequence[Optional[str]]]]),
+            "List[Set[Sequence[Optional[str]]]]",
+            id="nested-optional-sequence-of-primitives",
+        ),
+        pytest.param((List[Class1]), "List[Class1]", id="shallow-sequence-of-classes"),
+        pytest.param((List[Set[Sequence[Class1]]]), "List[Set[Sequence[Class1]]]", id="nested-sequence-of-classes"),
+        pytest.param((Dict[(str, int)]), "Dict[str, int]", id="shallow-mapping-of-primitives"),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            "Dict[str, Mapping[str, Dict[str, int]]]",
+            id="nested-mapping-of-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(Any, Dict[(str, int)])])]),
+            "Dict[str, Mapping[Any, Dict[str, int]]]",
+            id="nested-mapping-of-primitives-with-any",
+        ),
+        pytest.param((Dict[(str, Class1)]), "Dict[str, Class1]", id="shallow-mapping-of-classes"),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            "Dict[str, Mapping[str, Dict[str, Class1]]]",
+            id="nested-mapping-of-classes",
+        ),
+        pytest.param((Literal["a", "b", "c"]), "Literal['a', 'b', 'c']", id="string-literal"),
+        pytest.param((Literal[1, 2, 3]), "Literal[1, 2, 3]", id="primitive-literal"),
+        pytest.param((Literal[Enum1.TEST1]), "Literal[Enum1.TEST1]", id="enum-literal"),
+        pytest.param(
+            (Tuple[(Optional[Literal["a", "b", "c"]], Union[(Path, Dict[(int, Class1)])])]),
+            "Tuple[Optional[Literal['a', 'b', 'c']], Union[Path, Dict[int, Class1]]]",
+            id="deeply-nested-complex-type",
+        ),
+    ],
+)
+def test_type_name(type_, repr):
+    assert _type_name(type_) == repr
+
+
+@pytest.mark.parametrize(
+    "sender_type,receiver_type",
+    [
+        pytest.param(str, str, id="same-primitives"),
+        pytest.param(str, (Optional[str]), id="receiving-primitive-is-optional"),
+        pytest.param(str, (Union[(int, str)]), id="receiving-type-is-union-of-primitives"),
+        pytest.param((Union[(int, str)]), (Union[(int, str)]), id="identical-unions"),
+        pytest.param((Union[(int, str)]), (Union[(int, str, bool)]), id="receiving-union-is-superset-of-sender"),
+        pytest.param(str, Any, id="primitive-to-any"),
+        pytest.param(Class1, Class1, id="same-class"),
+        pytest.param(Class1, (Optional[Class1]), id="receiving-class-is-optional"),
+        pytest.param(Class1, Class1, id="class-to-any"),
+        pytest.param(Class3, Class1, id="subclass-to-class"),
+        pytest.param(Class1, (Union[(int, Class1)]), id="receiving-type-is-union-of-classes"),
+        pytest.param(Class3, (Union[(int, Class1)]), id="receiving-type-is-union-of-superclasses"),
+        pytest.param((List[int]), (List[int]), id="same-lists"),
+        pytest.param((List[int]), (Optional[List[int]]), id="receiving-list-is-optional"),
+        pytest.param((List[int]), (List[Any]), id="list-of-primitive-to-list-of-any"),
+        pytest.param((List[Class1]), (List[Class1]), id="list-of-same-classes"),
+        pytest.param((List[Class3]), (List[Class1]), id="list-of-subclass-to-list-of-class"),
+        pytest.param((List[Class1]), (List[Any]), id="list-of-classes-to-list-of-any"),
+        pytest.param(
+            (List[Set[Sequence[bool]]]), (List[Set[Sequence[bool]]]), id="nested-sequences-of-same-primitives"
+        ),
+        pytest.param(
+            (List[Set[Sequence[bool]]]),
+            (List[Set[Sequence[Any]]]),
+            id="nested-sequences-of-primitives-to-nested-sequences-of-any",
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class1]]]), (List[Set[Sequence[Class1]]]), id="nested-sequences-of-same-classes"
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class3]]]),
+            (List[Set[Sequence[Class1]]]),
+            id="nested-sequences-of-subclasses-to-nested-sequences-of-classes",
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class1]]]),
+            (List[Set[Sequence[Any]]]),
+            id="nested-sequences-of-classes-to-nested-sequences-of-any",
+        ),
+        pytest.param((Dict[(str, int)]), (Dict[(str, int)]), id="same-dicts-of-primitives"),
+        pytest.param((Dict[(str, int)]), (Dict[(Any, int)]), id="dict-of-primitives-to-dict-of-any-keys"),
+        pytest.param((Dict[(str, int)]), (Dict[(str, Any)]), id="dict-of-primitives-to-dict-of-any-values"),
+        pytest.param((Dict[(str, int)]), (Dict[(Any, Any)]), id="dict-of-primitives-to-dict-of-any-key-and-values"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(str, Class1)]), id="same-dicts-of-classes-values"),
+        pytest.param((Dict[(str, Class3)]), (Dict[(str, Class1)]), id="dict-of-subclasses-to-dict-of-classes"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(Any, Class1)]), id="dict-of-classes-to-dict-of-any-keys"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(str, Any)]), id="dict-of-classes-to-dict-of-any-values"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(Any, Any)]), id="dict-of-classes-to-dict-of-any-key-and-values"),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            id="nested-mappings-of-same-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(Any, int)])])]),
+            id="nested-mapping-of-primitives-to-nested-mapping-of-any-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(Any, Dict[(str, int)])])]),
+            id="nested-mapping-of-primitives-to-nested-mapping-of-higher-level-any-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Any)])])]),
+            id="nested-mapping-of-primitives-to-nested-mapping-of-any-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(Any, Dict[(Any, Any)])])]),
+            id="nested-mapping-of-primitives-to-nested-mapping-of-any-keys-and-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            id="nested-mappings-of-same-classes",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class3)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            id="nested-mapping-of-subclasses-to-nested-mapping-of-classes",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(str, Dict[(Any, Class1)])])]),
+            id="nested-mapping-of-classes-to-nested-mapping-of-any-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(Any, Dict[(str, Class1)])])]),
+            id="nested-mapping-of-classes-to-nested-mapping-of-higher-level-any-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Any)])])]),
+            id="nested-mapping-of-classes-to-nested-mapping-of-any-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(Any, Dict[(Any, Any)])])]),
+            id="nested-mapping-of-classes-to-nested-mapping-of-any-keys-and-values",
+        ),
+        pytest.param((Literal["a", "b", "c"]), (Literal["a", "b", "c"]), id="same-primitive-literal"),
+        pytest.param((Literal[Enum1.TEST1]), (Literal[Enum1.TEST1]), id="same-enum-literal"),
+        pytest.param(
+            (Tuple[(Optional[Literal["a", "b", "c"]], Union[(Path, Dict[(int, Class1)])])]),
+            (Tuple[(Optional[Literal["a", "b", "c"]], Union[(Path, Dict[(int, Class1)])])]),
+            id="identical-deeply-nested-complex-type",
+        ),
+    ],
+)
+def test_types_are_compatible(sender_type, receiver_type):
+    assert _types_are_compatible(sender_type, receiver_type)
+
+
+@pytest.mark.parametrize(
+    "sender_type,receiver_type",
+    [
+        pytest.param(int, bool, id="different-primitives"),
+        pytest.param(Class1, Class2, id="different-classes"),
+        pytest.param(Class1, Class3, id="class-to-subclass"),
+        pytest.param(Any, int, id="any-to-primitive"),
+        pytest.param(Any, Class2, id="any-to-class"),
+        pytest.param((Optional[str]), str, id="sending-primitive-is-optional"),
+        pytest.param((Optional[Class1]), Class1, id="sending-class-is-optional"),
+        pytest.param((Optional[List[int]]), (List[int]), id="sending-list-is-optional"),
+        pytest.param((Union[(int, str)]), str, id="sending-type-is-union"),
+        pytest.param((Union[(int, str, bool)]), (Union[(int, str)]), id="sending-union-is-superset-of-receiver"),
+        pytest.param((Union[(int, bool)]), (Union[(int, str)]), id="partially-overlapping-unions-with-primitives"),
+        pytest.param((Union[(int, Class1)]), (Union[(int, Class2)]), id="partially-overlapping-unions-with-classes"),
+        pytest.param((List[int]), (List[str]), id="different-lists-of-primitives"),
+        pytest.param((List[int]), List, id="list-of-primitive-to-bare-list"),
+        pytest.param((List[int]), list, id="list-of-primitive-to-list-object"),
+        pytest.param((List[Class1]), (List[Class2]), id="different-lists-of-classes"),
+        pytest.param((List[Class1]), (List[Class3]), id="lists-of-classes-to-subclasses"),
+        pytest.param((List[Any]), (List[str]), id="list-of-any-to-list-of-primitives"),
+        pytest.param((List[Any]), (List[Class2]), id="list-of-any-to-list-of-classes"),
+        pytest.param(
+            (List[Set[Sequence[str]]]), (List[Set[Sequence[bool]]]), id="nested-sequences-of-different-primitives"
+        ),
+        pytest.param(
+            (List[Set[Sequence[str]]]), (Set[List[Sequence[str]]]), id="different-nested-sequences-of-same-primitives"
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class1]]]), (List[Set[Sequence[Class2]]]), id="nested-sequences-of-different-classes"
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class1]]]), (List[Set[Sequence[Class3]]]), id="nested-sequences-of-classes-to-subclasses"
+        ),
+        pytest.param(
+            (List[Set[Sequence[Class1]]]), (Set[List[Sequence[Class1]]]), id="different-nested-sequences-of-same-class"
+        ),
+        pytest.param(
+            (List[Set[Sequence[Any]]]),
+            (List[Set[Sequence[bool]]]),
+            id="nested-list-of-Any-to-nested-list-of-primitives",
+        ),
+        pytest.param(
+            (List[Set[Sequence[Any]]]), (List[Set[Sequence[Class2]]]), id="nested-list-of-Any-to-nested-list-of-classes"
+        ),
+        pytest.param((Dict[(str, int)]), (Dict[(int, int)]), id="different-dict-of-primitive-keys"),
+        pytest.param((Dict[(str, int)]), (Dict[(str, bool)]), id="different-dict-of-primitive-values"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(str, Class2)]), id="different-dict-of-class-values"),
+        pytest.param((Dict[(str, Class1)]), (Dict[(str, Class3)]), id="different-dict-of-class-to-subclass-values"),
+        pytest.param((Dict[(Any, int)]), (Dict[(int, int)]), id="dict-of-Any-keys-to-dict-of-primitives"),
+        pytest.param((Dict[(str, Any)]), (Dict[(int, int)]), id="dict-of-Any-values-to-dict-of-primitives"),
+        pytest.param((Dict[(str, Any)]), (Dict[(int, Class1)]), id="dict-of-Any-values-to-dict-of-classes"),
+        pytest.param((Dict[(Any, Any)]), (Dict[(int, int)]), id="dict-of-Any-keys-and-values-to-dict-of-primitives"),
+        pytest.param((Dict[(Any, Any)]), (Dict[(int, Class1)]), id="dict-of-Any-keys-and-values-to-dict-of-classes"),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Mapping[(str, Dict[(str, Dict[(str, int)])])]),
+            id="different-nested-mappings-of-same-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(int, int)])])]),
+            id="same-nested-mappings-of-different-primitive-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(int, Dict[(str, int)])])]),
+            id="same-nested-mappings-of-different-higer-level-primitive-keys",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, bool)])])]),
+            id="same-nested-mappings-of-different-primitive-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class2)])])]),
+            id="same-nested-mappings-of-different-class-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class2)])])]),
+            id="same-nested-mappings-of-class-to-subclass-values",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(Any, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            id="nested-mapping-of-Any-keys-to-nested-mapping-of-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(Any, Dict[(Any, int)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            id="nested-mapping-of-higher-level-Any-keys-to-nested-mapping-of-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Any)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            id="nested-mapping-of-Any-values-to-nested-mapping-of-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(str, Any)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            id="nested-mapping-of-Any-values-to-nested-mapping-of-classes",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(Any, Any)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, int)])])]),
+            id="nested-mapping-of-Any-keys-and-values-to-nested-mapping-of-primitives",
+        ),
+        pytest.param(
+            (Dict[(str, Mapping[(str, Dict[(Any, Any)])])]),
+            (Dict[(str, Mapping[(str, Dict[(str, Class1)])])]),
+            id="nested-mapping-of-Any-keys-and-values-to-nested-mapping-of-classes",
+        ),
+        pytest.param((Literal["a", "b", "c"]), (Literal["x", "y"]), id="different-literal-of-same-primitive"),
+        pytest.param((Literal["a", "b", "c"]), (Literal["a", "b"]), id="subset-literal"),
+        pytest.param((Literal[Enum1.TEST1]), (Literal[Enum1.TEST2]), id="different-literal-of-same-enum"),
+        pytest.param(
+            (Tuple[(Optional[Literal["a", "b", "c"]], Union[(Path, Dict[(int, Class1)])])]),
+            (Tuple[(Literal["a", "b", "c"], Union[(Path, Dict[(int, Class1)])])]),
+            id="deeply-nested-complex-type-is-compatible-but-cannot-be-checked",
+        ),
+    ],
+)
+def test_types_are_not_compatible(sender_type, receiver_type):
+    assert not _types_are_compatible(sender_type, receiver_type)
diff --git a/test/core/pipeline/test_validation_pipeline_io.py b/test/core/pipeline/test_validation_pipeline_io.py
index 47fb4c59..f9160799 100644
--- a/test/core/pipeline/test_validation_pipeline_io.py
+++ b/test/core/pipeline/test_validation_pipeline_io.py
@@ -5,8 +5,7 @@ from typing import Optional
 
 import pytest
 
-from haystack.core.component.sockets import InputSocket, OutputSocket
-from haystack.core.component.types import Variadic
+from haystack.core.component.types import InputSocket, OutputSocket, Variadic
 from haystack.core.errors import PipelineValidationError
 from haystack.core.pipeline import Pipeline
 from haystack.core.pipeline.descriptions import find_pipeline_inputs, find_pipeline_outputs
diff --git a/test/core/test_utils.py b/test/core/test_utils.py
deleted file mode 100644
index 96af1435..00000000
--- a/test/core/test_utils.py
+++ /dev/null
@@ -1,79 +0,0 @@
-from typing import List, Set, Sequence, Tuple, Dict, Mapping, Literal, Union, Optional, Any
-from enum import Enum
-from pathlib import Path
-
-import pytest
-
-from haystack.core.type_utils import _type_name
-
-
-class Class1:
-    ...
-
-
-class Class2:
-    ...
-
-
-class Class3(Class1):
-    ...
-
-
-class Enum1(Enum):
-    TEST1 = Class1
-    TEST2 = Class2
-
-
-@pytest.mark.parametrize(
-    "type_,repr",
-    [
-        pytest.param(str, "str", id="primitive-types"),
-        pytest.param(Any, "Any", id="any"),
-        pytest.param(Class1, "Class1", id="class"),
-        pytest.param(Optional[int], "Optional[int]", id="shallow-optional-with-primitive"),
-        pytest.param(Optional[Any], "Optional[Any]", id="shallow-optional-with-any"),
-        pytest.param(Optional[Class1], "Optional[Class1]", id="shallow-optional-with-class"),
-        pytest.param(Union[bool, Class1], "Union[bool, Class1]", id="shallow-union"),
-        pytest.param(List[str], "List[str]", id="shallow-sequence-of-primitives"),
-        pytest.param(List[Set[Sequence[str]]], "List[Set[Sequence[str]]]", id="nested-sequence-of-primitives"),
-        pytest.param(
-            Optional[List[Set[Sequence[str]]]],
-            "Optional[List[Set[Sequence[str]]]]",
-            id="optional-nested-sequence-of-primitives",
-        ),
-        pytest.param(
-            List[Set[Sequence[Optional[str]]]],
-            "List[Set[Sequence[Optional[str]]]]",
-            id="nested-optional-sequence-of-primitives",
-        ),
-        pytest.param(List[Class1], "List[Class1]", id="shallow-sequence-of-classes"),
-        pytest.param(List[Set[Sequence[Class1]]], "List[Set[Sequence[Class1]]]", id="nested-sequence-of-classes"),
-        pytest.param(Dict[str, int], "Dict[str, int]", id="shallow-mapping-of-primitives"),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, int]]],
-            "Dict[str, Mapping[str, Dict[str, int]]]",
-            id="nested-mapping-of-primitives",
-        ),
-        pytest.param(
-            Dict[str, Mapping[Any, Dict[str, int]]],
-            "Dict[str, Mapping[Any, Dict[str, int]]]",
-            id="nested-mapping-of-primitives-with-any",
-        ),
-        pytest.param(Dict[str, Class1], "Dict[str, Class1]", id="shallow-mapping-of-classes"),
-        pytest.param(
-            Dict[str, Mapping[str, Dict[str, Class1]]],
-            "Dict[str, Mapping[str, Dict[str, Class1]]]",
-            id="nested-mapping-of-classes",
-        ),
-        pytest.param(Literal["a", "b", "c"], "Literal['a', 'b', 'c']", id="string-literal"),
-        pytest.param(Literal[1, 2, 3], "Literal[1, 2, 3]", id="primitive-literal"),
-        pytest.param(Literal[Enum1.TEST1], "Literal[Enum1.TEST1]", id="enum-literal"),
-        pytest.param(
-            Tuple[Optional[Literal["a", "b", "c"]], Union[Path, Dict[int, Class1]]],
-            "Tuple[Optional[Literal['a', 'b', 'c']], Union[Path, Dict[int, Class1]]]",
-            id="deeply-nested-complex-type",
-        ),
-    ],
-)
-def test_type_name(type_, repr):
-    assert _type_name(type_) == repr