Repository: deepset-ai/haystack

Issue:
JSON Convertor :  Pipeline Component
**Is your feature request related to a problem? Please describe.**
Currently we have a `.txt` to `Document` convertor besides others and unstructured. But I see most of the data we deal with are in the form of JSON.

**Describe the solution you'd like**
So a `.json` to `Document` convertor will be a bread winner while consuming API data in pipelines.

**Describe alternatives you've considered**
[Unstructured file convertor](https://docs.haystack.deepset.ai/docs/unstructuredfileconverter) is present but JSON schema as a individual convertor adds more sense and value.


Discussion:

Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9b599aaf..483ed06b 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -26,6 +26,7 @@ repos:
     rev: v2.3.0
     hooks:
       - id: codespell
+        args: ["--toml", "pyproject.toml"]
         additional_dependencies:
           - tomli
 
diff --git a/docs/pydoc/config/converters_api.yml b/docs/pydoc/config/converters_api.yml
index 945ae374..6c89138c 100644
--- a/docs/pydoc/config/converters_api.yml
+++ b/docs/pydoc/config/converters_api.yml
@@ -4,17 +4,18 @@ loaders:
     modules:
       [
         "azure",
+        "csv",
+        "docx",
         "html",
+        "json",
         "markdown",
+        "openapi_functions",
+        "output_adapter",
         "pdfminer",
-        "pypdf",
         "pptx",
+        "pypdf",
         "tika",
         "txt",
-        "output_adapter",
-        "openapi_functions",
-        "docx",
-        "csv"
       ]
     ignore_when_discovered: ["__init__"]
 processors:
diff --git a/haystack/components/converters/__init__.py b/haystack/components/converters/__init__.py
index 681ab85c..4561dd1e 100644
--- a/haystack/components/converters/__init__.py
+++ b/haystack/components/converters/__init__.py
@@ -6,6 +6,7 @@ from haystack.components.converters.azure import AzureOCRDocumentConverter
 from haystack.components.converters.csv import CSVToDocument
 from haystack.components.converters.docx import DOCXMetadata, DOCXToDocument
 from haystack.components.converters.html import HTMLToDocument
+from haystack.components.converters.json import JSONConverter
 from haystack.components.converters.markdown import MarkdownToDocument
 from haystack.components.converters.openapi_functions import OpenAPIServiceToFunctions
 from haystack.components.converters.output_adapter import OutputAdapter
@@ -29,4 +30,5 @@ __all__ = [
     "DOCXMetadata",
     "PPTXToDocument",
     "CSVToDocument",
+    "JSONConverter",
 ]
diff --git a/haystack/components/converters/json.py b/haystack/components/converters/json.py
new file mode 100644
index 00000000..966da881
--- /dev/null
+++ b/haystack/components/converters/json.py
@@ -0,0 +1,277 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from pathlib import Path
+from typing import Any, Dict, List, Literal, Optional, Set, Tuple, Union
+
+from haystack import component, default_from_dict, default_to_dict, logging
+from haystack.components.converters.utils import get_bytestream_from_source, normalize_metadata
+from haystack.dataclasses import ByteStream, Document
+from haystack.lazy_imports import LazyImport
+
+logger = logging.getLogger(__name__)
+
+with LazyImport("Run 'pip install jq'") as jq_import:
+    import jq
+
+
+@component
+class JSONConverter:
+    """
+    Converts one or more JSON files into a text document.
+
+    ### Usage examples
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    source = ByteStream.from_string(json.dumps({"text": "This is the content of my document"}))
+
+    converter = JSONConverter(content_key="text")
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'This is the content of my document'
+    ```
+
+    Optionally, you can also provide a `jq_schema` string to filter the JSON source files and `extra_meta_fields`
+    to extract from the filtered data:
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    data = {
+        "laureates": [
+            {
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced "
+                "by neutron irradiation, and for his related discovery of nuclear reactions brought about by"
+                " slow neutrons",
+            },
+            {
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+            },
+        ],
+    }
+    source = ByteStream.from_string(json.dumps(data))
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'for his demonstrations of the existence of new radioactive elements produced by
+    # neutron irradiation, and for his related discovery of nuclear reactions brought
+    # about by slow neutrons'
+
+    print(documents[0].meta)
+    # {'firstname': 'Enrico', 'surname': 'Fermi'}
+
+    print(documents[1].content)
+    # 'for their discoveries of growth factors'
+
+    print(documents[1].meta)
+    # {'firstname': 'Rita', 'surname': 'Levi-Montalcini'}
+    ```
+
+    """
+
+    def __init__(
+        self,
+        jq_schema: Optional[str] = None,
+        content_key: Optional[str] = None,
+        extra_meta_fields: Optional[Union[Set[str], Literal["*"]]] = None,
+    ):
+        """
+        Creates a JSONConverter component.
+
+        An optional `jq_schema` can be provided to extract nested data in the JSON source files.
+        See the [official jq documentation](https://jqlang.github.io/jq/) for more info on the filters syntax.
+        If `jq_schema` is not set, whole JSON source files will be used to extract content.
+
+        Optionally, you can provide a `content_key` to specify which key in the extracted object must
+        be set as the document's content.
+
+        If both `jq_schema` and `content_key` are set, the component will search for the `content_key` in
+        the JSON object extracted by `jq_schema`. If the extracted data is not a JSON object, it will be skipped.
+
+        If only `jq_schema` is set, the extracted data must be a scalar value. If it's a JSON object or array,
+        it will be skipped.
+
+        If only `content_key` is set, the source JSON file must be a JSON object, else it will be skipped.
+
+        `extra_meta_fields` can either be set to a set of strings or a literal `"*"` string.
+        If it's a set of strings, it must specify fields in the extracted objects that must be set in
+        the extracted documents. If a field is not found, the meta value will be `None`.
+        If set to `"*"`, all fields that are not `content_key` found in the filtered JSON object will
+        be saved as metadata.
+
+        Initialization will fail if neither `jq_schema` nor `content_key` are set.
+
+        :param jq_schema:
+            Optional jq filter string to extract content.
+            If not specified, whole JSON object will be used to extract information.
+        :param content_key:
+            Optional key to extract document content.
+            If `jq_schema` is specified, the `content_key` will be extracted from that object.
+        :param extra_meta_fields:
+            An optional set of meta keys to extract from the content.
+            If `jq_schema` is specified, all keys will be extracted from that object.
+        """
+        self._compiled_filter = None
+        if jq_schema:
+            jq_import.check()
+            self._compiled_filter = jq.compile(jq_schema)
+
+        self._jq_schema = jq_schema
+        self._content_key = content_key
+        self._meta_fields = extra_meta_fields
+
+        if self._compiled_filter is None and self._content_key is None:
+            msg = "No `jq_schema` nor `content_key` specified. Set either or both to extract data."
+            raise ValueError(msg)
+
+    def to_dict(self) -> Dict[str, Any]:
+        """
+        Serializes the component to a dictionary.
+
+        :returns:
+            Dictionary with serialized data.
+        """
+        return default_to_dict(
+            self, jq_schema=self._jq_schema, content_key=self._content_key, extra_meta_fields=self._meta_fields
+        )
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "JSONConverter":
+        """
+        Deserializes the component from a dictionary.
+
+        :param data:
+            Dictionary to deserialize from.
+        :returns:
+            Deserialized component.
+        """
+        return default_from_dict(cls, data)
+
+    def _get_content_and_meta(self, source: ByteStream) -> List[Tuple[str, Dict[str, Any]]]:
+        """
+        Utility function to extract text and metadata from a JSON file.
+
+        :param source:
+            UTF-8 byte stream.
+        :returns:
+            Collection of text and metadata dict tuples, each corresponding
+            to a different document.
+        """
+        try:
+            file_content = source.data.decode("utf-8")
+        except UnicodeError as exc:
+            logger.warning(
+                "Failed to extract text from {source}. Skipping it. Error: {error}",
+                source=source.meta["file_path"],
+                error=exc,
+            )
+
+        meta_fields = self._meta_fields or set()
+
+        if self._compiled_filter is not None:
+            try:
+                objects = list(self._compiled_filter.input_text(file_content))
+            except Exception as exc:
+                logger.warning(
+                    "Failed to extract text from {source}. Skipping it. Error: {error}",
+                    source=source.meta["file_path"],
+                    error=exc,
+                )
+                return []
+        else:
+            # We just load the whole file as JSON if the user didn't provide a jq filter.
+            # We put it in a list even if it's not to ease handling it later on.
+            objects = [json.loads(file_content)]
+
+        result = []
+        if self._content_key is not None:
+            for obj in objects:
+                if not isinstance(obj, dict):
+                    logger.warning("Expected a dictionary but got {obj}. Skipping it.", obj=obj)
+                    continue
+                if self._content_key not in obj:
+                    logger.warning(
+                        "'{content_key}' not found in {obj}. Skipping it.", content_key=self._content_key, obj=obj
+                    )
+                    continue
+
+                text = obj[self._content_key]
+                if isinstance(text, (dict, list)):
+                    logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
+                    continue
+
+                meta = {}
+                if meta_fields == "*":
+                    meta = {k: v for k, v in obj.items() if k != self._content_key}
+                else:
+                    for field in meta_fields:
+                        meta[field] = obj.get(field, None)
+                result.append((text, meta))
+        else:
+            for obj in objects:
+                if isinstance(obj, (dict, list)):
+                    logger.warning("Expected a scalar value but got {obj}. Skipping it.", obj=obj)
+                    continue
+                result.append((str(obj), {}))
+
+        return result
+
+    @component.output_types(documents=List[Document])
+    def run(
+        self,
+        sources: List[Union[str, Path, ByteStream]],
+        meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
+    ):
+        """
+        Converts a list of JSON files to documents.
+
+        :param sources:
+            A list of file paths or ByteStream objects.
+        :param meta:
+            Optional metadata to attach to the documents.
+            This value can be either a list of dictionaries or a single dictionary.
+            If it's a single dictionary, its content is added to the metadata of all produced documents.
+            If it's a list, the length of the list must match the number of sources.
+            If `sources` contain ByteStream objects, their `meta` will be added to the output documents.
+
+        :returns:
+            A dictionary with the following keys:
+            - `documents`: A list of created documents.
+        """
+        documents = []
+        meta_list = normalize_metadata(meta=meta, sources_count=len(sources))
+
+        for source, metadata in zip(sources, meta_list):
+            try:
+                bytestream = get_bytestream_from_source(source)
+            except Exception as exc:
+                logger.warning("Could not read {source}. Skipping it. Error: {error}", source=source, error=exc)
+                continue
+
+            data = self._get_content_and_meta(bytestream)
+
+            for text, extra_meta in data:
+                merged_metadata = {**bytestream.meta, **metadata, **extra_meta}
+                document = Document(content=text, meta=merged_metadata)
+                documents.append(document)
+
+        return {"documents": documents}
diff --git a/haystack/components/joiners/branch.py b/haystack/components/joiners/branch.py
index 788d78a9..49404ac4 100644
--- a/haystack/components/joiners/branch.py
+++ b/haystack/components/joiners/branch.py
@@ -5,13 +5,13 @@
 from typing import Any, Dict, Type
 
 from haystack import component, default_from_dict, default_to_dict, logging
-from haystack.core.component.types import Variadic
+from haystack.core.component.types import GreedyVariadic
 from haystack.utils import deserialize_type, serialize_type
 
 logger = logging.getLogger(__name__)
 
 
-@component(is_greedy=True)
+@component()
 class BranchJoiner:
     """
     A component to join different branches of a pipeline into one single output.
@@ -100,7 +100,7 @@ class BranchJoiner:
         """
         self.type_ = type_
         # type_'s type can't be determined statically
-        component.set_input_types(self, value=Variadic[type_])  # type: ignore
+        component.set_input_types(self, value=GreedyVariadic[type_])  # type: ignore
         component.set_output_types(self, value=type_)
 
     def to_dict(self):
diff --git a/haystack/core/component/component.py b/haystack/core/component/component.py
index 72c444fd..e1dbf2c5 100644
--- a/haystack/core/component/component.py
+++ b/haystack/core/component/component.py
@@ -71,6 +71,7 @@ method decorated with `@component.input`. This dataclass contains:
 
 import inspect
 import sys
+import warnings
 from collections.abc import Callable
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -292,17 +293,6 @@ class ComponentMeta(type):
         # We use this flag to check that.
         instance.__haystack_added_to_pipeline__ = None
 
-        # Only Components with variadic inputs can be greedy. If the user set the greedy flag
-        # to True, but the component doesn't have a variadic input, we set it to False.
-        # We can have this information only at instance creation time, so we do it here.
-        is_variadic = any(socket.is_variadic for socket in instance.__haystack_input__._sockets_dict.values())
-        if not is_variadic and cls.__haystack_is_greedy__:
-            logger.warning(
-                "Component '{component}' has no variadic input, but it's marked as greedy. "
-                "This is not supported and can lead to unexpected behavior.",
-                component=cls.__name__,
-            )
-
         return instance
 
 
@@ -497,12 +487,21 @@ class _Component:
 
         return output_types_decorator
 
-    def _component(self, cls, is_greedy: bool = False):
+    def _component(self, cls, is_greedy: Optional[bool] = None):
         """
         Decorator validating the structure of the component and registering it in the components registry.
         """
         logger.debug("Registering {component} as a component", component=cls)
 
+        if is_greedy is not None:
+            msg = (
+                "The 'is_greedy' argument is deprecated and will be removed in version '2.7.0'. "
+                "Change the 'Variadic' input of your Component to 'GreedyVariadic' instead."
+            )
+            warnings.warn(msg, DeprecationWarning)
+        else:
+            is_greedy = False
+
         # Check for required methods and fail as soon as possible
         if not hasattr(cls, "run"):
             raise ComponentError(f"{cls.__name__} must have a 'run()' method. See the docs for more information.")
@@ -542,15 +541,9 @@ class _Component:
         # Override the __repr__ method with a default one
         cls.__repr__ = _component_repr
 
-        # The greedy flag can be True only if the component has a variadic input.
-        # At this point of the lifetime of the component, we can't reliably know if it has a variadic input.
-        # So we set it to whatever the user specified, during the instance creation we'll change it if needed
-        # since we'll have access to the input sockets and check if any of them is variadic.
-        setattr(cls, "__haystack_is_greedy__", is_greedy)
-
         return cls
 
-    def __call__(self, cls: Optional[type] = None, is_greedy: bool = False):
+    def __call__(self, cls: Optional[type] = None, is_greedy: Optional[bool] = None):
         # We must wrap the call to the decorator in a function for it to work
         # correctly with or without parens
         def wrap(cls):
diff --git a/haystack/core/component/types.py b/haystack/core/component/types.py
index f7aa8d08..b08681a7 100644
--- a/haystack/core/component/types.py
+++ b/haystack/core/component/types.py
@@ -8,6 +8,7 @@ from typing import Any, Iterable, List, Type, TypeVar, get_args
 from typing_extensions import Annotated, TypeAlias  # Python 3.8 compatibility
 
 HAYSTACK_VARIADIC_ANNOTATION = "__haystack__variadic_t"
+HAYSTACK_GREEDY_VARIADIC_ANNOTATION = "__haystack__greedy_variadic_t"
 
 # # Generic type variable used in the Variadic container
 T = TypeVar("T")
@@ -16,9 +17,17 @@ T = TypeVar("T")
 # Variadic is a custom annotation type we use to mark input types.
 # This type doesn't do anything else than "marking" the contained
 # type so it can be used in the `InputSocket` creation where we
-# check that its annotation equals to CANALS_VARIADIC_ANNOTATION
+# check that its annotation equals to HAYSTACK_VARIADIC_ANNOTATION
 Variadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_VARIADIC_ANNOTATION]
 
+# GreedyVariadic type is similar to Variadic.
+# The only difference is the way it's treated by the Pipeline when input is received
+# in a socket with this type.
+# Instead of waiting for other inputs to be received, Components that have a GreedyVariadic
+# input will be run right after receiving the first input.
+# Even if there are multiple connections to that socket.
+GreedyVariadic: TypeAlias = Annotated[Iterable[T], HAYSTACK_GREEDY_VARIADIC_ANNOTATION]
+
 
 class _empty:
     """Custom object for marking InputSocket.default_value as not set."""
@@ -37,6 +46,8 @@ class InputSocket:
         The default value of the input. If not set, the input is mandatory.
     :param is_variadic:
         Whether the input is variadic or not.
+    :param is_greedy
+        Whether the input is a greedy variadic or not.
     :param senders:
         The list of components that send data to this input.
     """
@@ -45,6 +56,7 @@ class InputSocket:
     type: Type
     default_value: Any = _empty
     is_variadic: bool = field(init=False)
+    is_greedy: bool = field(init=False)
     senders: List[str] = field(default_factory=list)
 
     @property
@@ -55,9 +67,14 @@ class InputSocket:
     def __post_init__(self):
         try:
             # __metadata__ is a tuple
-            self.is_variadic = self.type.__metadata__[0] == HAYSTACK_VARIADIC_ANNOTATION
+            self.is_variadic = self.type.__metadata__[0] in [
+                HAYSTACK_VARIADIC_ANNOTATION,
+                HAYSTACK_GREEDY_VARIADIC_ANNOTATION,
+            ]
+            self.is_greedy = self.type.__metadata__[0] == HAYSTACK_GREEDY_VARIADIC_ANNOTATION
         except AttributeError:
             self.is_variadic = False
+            self.is_greedy = False
         if self.is_variadic:
             # We need to "unpack" the type inside the Variadic annotation,
             # otherwise the pipeline connection api will try to match
diff --git a/haystack/core/pipeline/base.py b/haystack/core/pipeline/base.py
index 7044c9a3..8bb6b571 100644
--- a/haystack/core/pipeline/base.py
+++ b/haystack/core/pipeline/base.py
@@ -428,7 +428,7 @@ class PipelineBase:
 
         return instance
 
-    def connect(self, sender: str, receiver: str) -> "PipelineBase":
+    def connect(self, sender: str, receiver: str) -> "PipelineBase":  # noqa: PLR0915
         """
         Connects two components together.
 
@@ -453,6 +453,10 @@ class PipelineBase:
         sender_component_name, sender_socket_name = parse_connect_string(sender)
         receiver_component_name, receiver_socket_name = parse_connect_string(receiver)
 
+        if sender_component_name == receiver_component_name:
+            msg = "Connecting a Component to itself is deprecated and will raise an error from version '2.7.0' onwards."
+            warnings.warn(msg, DeprecationWarning)
+
         # Get the nodes data.
         try:
             from_sockets = self.graph.nodes[sender_component_name]["output_sockets"]
@@ -978,9 +982,8 @@ class PipelineBase:
             receiver = self.graph.nodes[receiver_name]["instance"]
             pair = (receiver_name, receiver)
 
-            is_greedy = getattr(receiver, "__haystack_is_greedy__", False)
             if receiver_socket.is_variadic:
-                if is_greedy:
+                if receiver_socket.is_greedy:
                     # If the receiver is greedy, we can run it as soon as possible.
                     # First we remove it from the status lists it's in if it's there or
                     # we risk running it multiple times.
@@ -1214,7 +1217,7 @@ def _connections_status(
 
 def _is_lazy_variadic(c: Component) -> bool:
     """
-    Small utility function to check if a Component has a Variadic input that is not greedy
+    Small utility function to check if a Component has at least a Variadic input and no GreedyVariadic input.
     """
     is_variadic = any(
         socket.is_variadic
@@ -1222,7 +1225,10 @@ def _is_lazy_variadic(c: Component) -> bool:
     )
     if not is_variadic:
         return False
-    return not getattr(c, "__haystack_is_greedy__", False)
+    return not any(
+        socket.is_greedy
+        for socket in c.__haystack_input__._sockets_dict.values()  # type: ignore
+    )
 
 
 def _has_all_inputs_with_defaults(c: Component) -> bool:
diff --git a/haystack/core/pipeline/pipeline.py b/haystack/core/pipeline/pipeline.py
index b51aa881..62e7c8f3 100644
--- a/haystack/core/pipeline/pipeline.py
+++ b/haystack/core/pipeline/pipeline.py
@@ -77,7 +77,7 @@ class Pipeline(PipelineBase):
             if not isinstance(res, Mapping):
                 raise PipelineRuntimeError(
                     f"Component '{name}' didn't return a dictionary. "
-                    "Components must always return dictionaries: check the the documentation."
+                    "Components must always return dictionaries: check the documentation."
                 )
             span.set_tag("haystack.component.visits", self.graph.nodes[name]["visits"])
             span.set_content_tag("haystack.component.output", res)
diff --git a/haystack/core/pipeline/predefined/chat_with_website.yaml.jinja2 b/haystack/core/pipeline/predefined/chat_with_website.yaml.jinja2
index ef299474..c4ef14d4 100644
--- a/haystack/core/pipeline/predefined/chat_with_website.yaml.jinja2
+++ b/haystack/core/pipeline/predefined/chat_with_website.yaml.jinja2
@@ -1,8 +1,8 @@
 components:
   converter:
-    init_parameters:
-      extractor_type: DefaultExtractor
     type: haystack.components.converters.html.HTMLToDocument
+    init_parameters:
+      extraction_kwargs: null
 
   fetcher:
     init_parameters:
@@ -22,7 +22,7 @@ components:
         strict: true
         type: env_var
       generation_kwargs: {}
-      model: gpt-3.5-turbo
+      model: gpt-4o-mini
       streaming_callback: null
       system_prompt: null
     type: haystack.components.generators.openai.OpenAIGenerator
diff --git a/haystack/core/pipeline/predefined/generative_qa.yaml.jinja2 b/haystack/core/pipeline/predefined/generative_qa.yaml.jinja2
index 83ffd588..456689d7 100644
--- a/haystack/core/pipeline/predefined/generative_qa.yaml.jinja2
+++ b/haystack/core/pipeline/predefined/generative_qa.yaml.jinja2
@@ -7,7 +7,7 @@ components:
         env_vars: [ "OPENAI_API_KEY" ]
         strict: true
         type: "env_var"
-      model: "gpt-3.5-turbo"
+      model: "gpt-4o-mini"
     type: "haystack.components.generators.openai.OpenAIGenerator"
 
   prompt_builder:
diff --git a/haystack/core/pipeline/predefined/rag.yaml.jinja2 b/haystack/core/pipeline/predefined/rag.yaml.jinja2
index 97b9ae76..cff8bbb8 100644
--- a/haystack/core/pipeline/predefined/rag.yaml.jinja2
+++ b/haystack/core/pipeline/predefined/rag.yaml.jinja2
@@ -10,7 +10,7 @@ components:
         strict: true
         type: env_var
       generation_kwargs: {}
-      model: gpt-3.5-turbo
+      model: gpt-4o-mini
       streaming_callback: null
       system_prompt: null
     type: haystack.components.generators.openai.OpenAIGenerator
diff --git a/pyproject.toml b/pyproject.toml
index 5c75292f..be286052 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -102,8 +102,9 @@ extra-dependencies = [
   "trafilatura",                      # HTMLToDocument
   "python-pptx",                      # PPTXToDocument
   "python-docx",                      # DocxToDocument
+  "jq",                               #JSONConverter
 
-  "nltk",                             # NLTKDocumentSplitter
+  "nltk", # NLTKDocumentSplitter
 
   # OpenAPI
   "jsonref",  # OpenAPIServiceConnector, OpenAPIServiceToFunctions
diff --git a/releasenotes/notes/deprecate-greedy-argument-4b8c39572f5df25c.yaml b/releasenotes/notes/deprecate-greedy-argument-4b8c39572f5df25c.yaml
new file mode 100644
index 00000000..c868a9a1
--- /dev/null
+++ b/releasenotes/notes/deprecate-greedy-argument-4b8c39572f5df25c.yaml
@@ -0,0 +1,13 @@
+---
+enhancements:
+  - |
+    Add new `GreedyVariadic` input type. This has a similar behaviour to `Variadic` input type
+    as it can be connected to multiple output sockets, though the Pipeline will run it as soon
+    as it receives an input without waiting for others.
+    This replaces the `is_greedy` argument in the `@component` decorator.
+    If you had a Component with a `Variadic` input type and `@component(is_greedy=True)` you need
+    to change the type to `GreedyVariadic` and remove `is_greedy=true` from `@component`.
+deprecations:
+  - |
+    `@component` decorator `is_greedy` argument is deprecated and will be removed in version `2.7.0`.
+    Use `GreedyVariadic` type instead.
diff --git a/releasenotes/notes/deprecate-self-connection-665647384ae2792b.yaml b/releasenotes/notes/deprecate-self-connection-665647384ae2792b.yaml
new file mode 100644
index 00000000..7a19a292
--- /dev/null
+++ b/releasenotes/notes/deprecate-self-connection-665647384ae2792b.yaml
@@ -0,0 +1,4 @@
+---
+deprecations:
+  - |
+    Deprecate connecting a Component to itself when calling `Pipeline.connect()`, it will raise an error from version `2.7.0` onwards
diff --git a/releasenotes/notes/fix-chat-web-pipe-template-0b00144b1099cda3.yaml b/releasenotes/notes/fix-chat-web-pipe-template-0b00144b1099cda3.yaml
new file mode 100644
index 00000000..60b2f90a
--- /dev/null
+++ b/releasenotes/notes/fix-chat-web-pipe-template-0b00144b1099cda3.yaml
@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Update the `CHAT_WITH_WEBSITE` Pipeline template to reflect the changes in
+    the `HTMLToDocument` converter component.
diff --git a/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml b/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml
new file mode 100644
index 00000000..f8a269d2
--- /dev/null
+++ b/releasenotes/notes/json-converter-a747e9c44543dfb5.yaml
@@ -0,0 +1,48 @@
+---
+features:
+  - |
+    Add new `JSONConverter` Component to convert JSON files to `Document`.
+    Optionally it can use jq to filter the source JSON files and extract only specific parts.
+
+    ```python
+    import json
+
+    from haystack.components.converters import JSONConverter
+    from haystack.dataclasses import ByteStream
+
+    data = {
+        "laureates": [
+            {
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced "
+                "by neutron irradiation, and for his related discovery of nuclear reactions brought about by slow neutrons",
+            },
+            {
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+            },
+        ],
+    }
+    source = ByteStream.from_string(json.dumps(data))
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields=["firstname", "surname"]
+    )
+
+    results = converter.run(sources=[source])
+    documents = results["documents"]
+    print(documents[0].content)
+    # 'for his demonstrations of the existence of new radioactive elements produced by
+    # neutron irradiation, and for his related discovery of nuclear reactions brought
+    # about by slow neutrons'
+
+    print(documents[0].meta)
+    # {'firstname': 'Enrico', 'surname': 'Fermi'}
+
+    print(documents[1].content)
+    # 'for their discoveries of growth factors'
+
+    print(documents[1].meta)
+    # {'firstname': 'Rita', 'surname': 'Levi-Montalcini'}
+    ```


Test patch:
diff --git a/test/components/converters/test_json.py b/test/components/converters/test_json.py
new file mode 100644
index 00000000..fe43044f
--- /dev/null
+++ b/test/components/converters/test_json.py
@@ -0,0 +1,433 @@
+# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import json
+from unittest.mock import patch
+from pathlib import Path
+import logging
+
+import pytest
+
+from haystack.components.converters import JSONConverter
+from haystack.dataclasses import ByteStream
+
+
+test_data = [
+    {
+        "year": "1997",
+        "category": "literature",
+        "laureates": [
+            {
+                "id": "674",
+                "firstname": "Dario",
+                "surname": "Fo",
+                "motivation": "who emulates the jesters of the Middle Ages in scourging authority and upholding the dignity of the downtrodden",
+                "share": "1",
+            }
+        ],
+    },
+    {
+        "year": "1986",
+        "category": "medicine",
+        "laureates": [
+            {
+                "id": "434",
+                "firstname": "Stanley",
+                "surname": "Cohen",
+                "motivation": "for their discoveries of growth factors",
+                "share": "2",
+            },
+            {
+                "id": "435",
+                "firstname": "Rita",
+                "surname": "Levi-Montalcini",
+                "motivation": "for their discoveries of growth factors",
+                "share": "2",
+            },
+        ],
+    },
+    {
+        "year": "1938",
+        "category": "physics",
+        "laureates": [
+            {
+                "id": "46",
+                "firstname": "Enrico",
+                "surname": "Fermi",
+                "motivation": "for his demonstrations of the existence of new radioactive elements produced by neutron irradiation, and for his related discovery of nuclear reactions brought about by slow neutrons",
+                "share": "1",
+            }
+        ],
+    },
+]
+
+
+def test_init_without_jq_schema_and_content_key():
+    with pytest.raises(
+        ValueError, match="No `jq_schema` nor `content_key` specified. Set either or both to extract data."
+    ):
+        JSONConverter()
+
+
+@patch("haystack.components.converters.json.jq_import")
+def test_init_without_jq_schema_and_missing_dependency(jq_import):
+    converter = JSONConverter(content_key="foo")
+    jq_import.check.assert_not_called()
+    assert converter._jq_schema is None
+    assert converter._content_key == "foo"
+    assert converter._meta_fields is None
+
+
+@patch("haystack.components.converters.json.jq_import")
+def test_init_with_jq_schema_and_missing_dependency(jq_import):
+    jq_import.check.side_effect = ImportError
+    with pytest.raises(ImportError):
+        JSONConverter(jq_schema=".laureates[].motivation")
+
+
+def test_init_with_jq_schema():
+    converter = JSONConverter(jq_schema=".")
+    assert converter._jq_schema == "."
+    assert converter._content_key is None
+    assert converter._meta_fields is None
+
+
+def test_to_dict():
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+
+    assert converter.to_dict() == {
+        "type": "haystack.components.converters.json.JSONConverter",
+        "init_parameters": {
+            "content_key": "motivation",
+            "jq_schema": ".laureates[]",
+            "extra_meta_fields": {"firstname", "surname"},
+        },
+    }
+
+
+def test_from_dict():
+    data = {
+        "type": "haystack.components.converters.json.JSONConverter",
+        "init_parameters": {
+            "content_key": "motivation",
+            "jq_schema": ".laureates[]",
+            "extra_meta_fields": ["firstname", "surname"],
+        },
+    }
+    converter = JSONConverter.from_dict(data)
+
+    assert converter._jq_schema == ".laureates[]"
+    assert converter._content_key == "motivation"
+    assert converter._meta_fields == ["firstname", "surname"]
+
+
+def test_run(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file)}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {}
+
+
+def test_run_with_non_json_file(tmpdir, caplog):
+    test_file = Path(tmpdir / "test_file.md")
+    test_file.write_text("This is not a JSON file.", "utf-8")
+
+    sources = [test_file]
+    converter = JSONConverter(".laureates | .motivation")
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING):
+        result = converter.run(sources=sources)
+
+    records = caplog.records
+    assert len(records) == 1
+    assert (
+        records[0].msg
+        == f"Failed to extract text from {test_file}. Skipping it. Error: parse error: Invalid numeric literal at line 1, column 5"
+    )
+    assert result == {"documents": []}
+
+
+def test_run_with_bad_filter(tmpdir, caplog):
+    test_file = Path(tmpdir / "test_file.json")
+    test_file.write_text(json.dumps(test_data[0]), "utf-8")
+
+    sources = [test_file]
+    converter = JSONConverter(".laureates | .motivation")
+
+    caplog.clear()
+    with caplog.at_level(logging.WARNING):
+        result = converter.run(sources=sources)
+
+    records = caplog.records
+    assert len(records) == 1
+    assert (
+        records[0].msg
+        == f'Failed to extract text from {test_file}. Skipping it. Error: Cannot index array with string "motivation"'
+    )
+    assert result == {"documents": []}
+
+
+def test_run_with_single_meta(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    meta = {"creation_date": "1945-05-25T00:00:00"}
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources, meta=meta)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"creation_date": "1945-05-25T00:00:00"}
+
+
+def test_run_with_meta_list(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    meta = [
+        {"creation_date": "1945-05-25T00:00:00"},
+        {"creation_date": "1943-09-03T00:00:00"},
+        {"creation_date": "1989-11-09T00:00:00"},
+    ]
+    converter = JSONConverter(jq_schema='.laureates[] | .firstname + " " + .surname + " " + .motivation')
+    result = converter.run(sources=sources, meta=meta)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "Dario Fo who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "creation_date": "1945-05-25T00:00:00"}
+    assert result["documents"][1].content == "Stanley Cohen for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "creation_date": "1943-09-03T00:00:00"}
+    assert result["documents"][2].content == "Rita Levi-Montalcini for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file), "creation_date": "1943-09-03T00:00:00"}
+    assert (
+        result["documents"][3].content == "Enrico Fermi for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"creation_date": "1989-11-09T00:00:00"}
+
+
+def test_run_with_meta_list_of_differing_length(tmpdir):
+    sources = ["random_file.json"]
+
+    meta = [{}, {}]
+    converter = JSONConverter(jq_schema=".")
+    with pytest.raises(ValueError, match="The length of the metadata list must match the number of sources."):
+        converter.run(sources=sources, meta=meta)
+
+
+def test_run_with_jq_schema_and_content_key(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(jq_schema=".laureates[]", content_key="motivation")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content == "who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {"file_path": str(second_test_file)}
+    assert (
+        result["documents"][3].content == "for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {}
+
+
+def test_run_with_jq_schema_content_key_and_extra_meta_fields(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(
+        jq_schema=".laureates[]", content_key="motivation", extra_meta_fields={"firstname", "surname"}
+    )
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content == "who emulates the jesters of the Middle Ages in scourging authority and "
+        "upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "firstname": "Dario", "surname": "Fo"}
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {
+        "file_path": str(second_test_file),
+        "firstname": "Stanley",
+        "surname": "Cohen",
+    }
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {
+        "file_path": str(second_test_file),
+        "firstname": "Rita",
+        "surname": "Levi-Montalcini",
+    }
+    assert (
+        result["documents"][3].content == "for his demonstrations of the existence of new "
+        "radioactive elements produced by neutron irradiation, and for his related discovery of nuclear "
+        "reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"firstname": "Enrico", "surname": "Fermi"}
+
+
+def test_run_with_content_key(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(content_key="category")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 3
+    assert result["documents"][0].content == "literature"
+    assert result["documents"][0].meta == {"file_path": str(first_test_file)}
+    assert result["documents"][1].content == "medicine"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file)}
+    assert result["documents"][2].content == "physics"
+    assert result["documents"][2].meta == {}
+
+
+def test_run_with_content_key_and_extra_meta_fields(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(content_key="category", extra_meta_fields={"year"})
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 3
+    assert result["documents"][0].content == "literature"
+    assert result["documents"][0].meta == {"file_path": str(first_test_file), "year": "1997"}
+    assert result["documents"][1].content == "medicine"
+    assert result["documents"][1].meta == {"file_path": str(second_test_file), "year": "1986"}
+    assert result["documents"][2].content == "physics"
+    assert result["documents"][2].meta == {"year": "1938"}
+
+
+def test_run_with_jq_schema_content_key_and_extra_meta_fields_literal(tmpdir):
+    first_test_file = Path(tmpdir / "first_test_file.json")
+    second_test_file = Path(tmpdir / "second_test_file.json")
+
+    first_test_file.write_text(json.dumps(test_data[0]), "utf-8")
+    second_test_file.write_text(json.dumps(test_data[1]), "utf-8")
+    byte_stream = ByteStream.from_string(json.dumps(test_data[2]))
+
+    sources = [str(first_test_file), second_test_file, byte_stream]
+    converter = JSONConverter(jq_schema=".laureates[]", content_key="motivation", extra_meta_fields="*")
+    result = converter.run(sources=sources)
+    assert len(result) == 1
+    assert len(result["documents"]) == 4
+    assert (
+        result["documents"][0].content
+        == "who emulates the jesters of the Middle Ages in scourging authority and upholding the dignity of the downtrodden"
+    )
+    assert result["documents"][0].meta == {
+        "file_path": str(first_test_file),
+        "id": "674",
+        "firstname": "Dario",
+        "surname": "Fo",
+        "share": "1",
+    }
+    assert result["documents"][1].content == "for their discoveries of growth factors"
+    assert result["documents"][1].meta == {
+        "file_path": str(second_test_file),
+        "id": "434",
+        "firstname": "Stanley",
+        "surname": "Cohen",
+        "share": "2",
+    }
+    assert result["documents"][2].content == "for their discoveries of growth factors"
+    assert result["documents"][2].meta == {
+        "file_path": str(second_test_file),
+        "id": "435",
+        "firstname": "Rita",
+        "surname": "Levi-Montalcini",
+        "share": "2",
+    }
+    assert (
+        result["documents"][3].content
+        == "for his demonstrations of the existence of new radioactive elements produced by neutron irradiation, "
+        "and for his related discovery of nuclear reactions brought about by slow neutrons"
+    )
+    assert result["documents"][3].meta == {"id": "46", "firstname": "Enrico", "surname": "Fermi", "share": "1"}
diff --git a/test/core/component/test_component.py b/test/core/component/test_component.py
index 8b4266db..49d3a8bd 100644
--- a/test/core/component/test_component.py
+++ b/test/core/component/test_component.py
@@ -430,57 +430,6 @@ def test_repr_added_to_pipeline():
     assert repr(comp) == f"{object.__repr__(comp)}\nmy_component\nInputs:\n  - value: int\nOutputs:\n  - value: int"
 
 
-def test_is_greedy_default_with_variadic_input():
-    @component
-    class MockComponent:
-        @component.output_types(value=int)
-        def run(self, value: Variadic[int]):
-            return {"value": value}
-
-    assert not MockComponent.__haystack_is_greedy__
-    assert not MockComponent().__haystack_is_greedy__
-
-
-def test_is_greedy_default_without_variadic_input():
-    @component
-    class MockComponent:
-        @component.output_types(value=int)
-        def run(self, value: int):
-            return {"value": value}
-
-    assert not MockComponent.__haystack_is_greedy__
-    assert not MockComponent().__haystack_is_greedy__
-
-
-def test_is_greedy_flag_with_variadic_input():
-    @component(is_greedy=True)
-    class MockComponent:
-        @component.output_types(value=int)
-        def run(self, value: Variadic[int]):
-            return {"value": value}
-
-    assert MockComponent.__haystack_is_greedy__
-    assert MockComponent().__haystack_is_greedy__
-
-
-def test_is_greedy_flag_without_variadic_input(caplog):
-    caplog.set_level(logging.WARNING)
-
-    @component(is_greedy=True)
-    class MockComponent:
-        @component.output_types(value=int)
-        def run(self, value: int):
-            return {"value": value}
-
-    assert MockComponent.__haystack_is_greedy__
-    assert caplog.text == ""
-    assert MockComponent().__haystack_is_greedy__
-    assert (
-        "Component 'MockComponent' has no variadic input, but it's marked as greedy."
-        " This is not supported and can lead to unexpected behavior.\n" in caplog.text
-    )
-
-
 def test_pre_init_hooking():
     @component
     class MockComponent:
diff --git a/test/core/pipeline/features/test_run.py b/test/core/pipeline/features/test_run.py
index 0b3541c8..c1492c48 100644
--- a/test/core/pipeline/features/test_run.py
+++ b/test/core/pipeline/features/test_run.py
@@ -38,7 +38,7 @@ scenarios("pipeline_run.feature")
 
 @given("a pipeline that has no components", target_fixture="pipeline_data")
 def pipeline_that_has_no_components():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     inputs = {}
     expected_outputs = {}
     return pipeline, [PipelineRunData(inputs=inputs, expected_outputs=expected_outputs)]
@@ -46,7 +46,7 @@ def pipeline_that_has_no_components():
 
 @given("a pipeline that is linear", target_fixture="pipeline_data")
 def pipeline_that_is_linear():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_addition", AddFixedValue(add=2))
     pipeline.add_component("second_addition", AddFixedValue())
     pipeline.add_component("double", Double())
@@ -183,7 +183,7 @@ def pipeline_that_has_a_single_component_with_a_default_input():
         def run(self, a: int, b: int = 2):
             return {"c": a + b}
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("with_defaults", WithDefault())
 
     return (
@@ -353,7 +353,7 @@ def pipeline_that_has_a_single_loop_with_two_conditional_branches():
 
 @given("a pipeline that has a component with dynamic inputs defined in init", target_fixture="pipeline_data")
 def pipeline_that_has_a_component_with_dynamic_inputs_defined_in_init():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("hello", Hello())
     pipeline.add_component("fstring", FString(template="This is the greeting: {greeting}!", variables=["greeting"]))
     pipeline.add_component("splitter", TextSplitter())
@@ -379,7 +379,7 @@ def pipeline_that_has_a_component_with_dynamic_inputs_defined_in_init():
 
 @given("a pipeline that has two branches that don't merge", target_fixture="pipeline_data")
 def pipeline_that_has_two_branches_that_dont_merge():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("add_one", AddFixedValue(add=1))
     pipeline.add_component("parity", Parity())
     pipeline.add_component("add_ten", AddFixedValue(add=10))
@@ -410,7 +410,7 @@ def pipeline_that_has_two_branches_that_dont_merge():
 
 @given("a pipeline that has three branches that don't merge", target_fixture="pipeline_data")
 def pipeline_that_has_three_branches_that_dont_merge():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("add_one", AddFixedValue(add=1))
     pipeline.add_component("repeat", Repeat(outputs=["first", "second"]))
     pipeline.add_component("add_ten", AddFixedValue(add=10))
@@ -438,7 +438,7 @@ def pipeline_that_has_three_branches_that_dont_merge():
 
 @given("a pipeline that has two branches that merge", target_fixture="pipeline_data")
 def pipeline_that_has_two_branches_that_merge():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_addition", AddFixedValue(add=2))
     pipeline.add_component("second_addition", AddFixedValue(add=2))
     pipeline.add_component("third_addition", AddFixedValue(add=2))
@@ -465,7 +465,7 @@ def pipeline_that_has_two_branches_that_merge():
     "a pipeline that has different combinations of branches that merge and do not merge", target_fixture="pipeline_data"
 )
 def pipeline_that_has_different_combinations_of_branches_that_merge_and_do_not_merge():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("add_one", AddFixedValue())
     pipeline.add_component("parity", Parity())
     pipeline.add_component("add_ten", AddFixedValue(add=10))
@@ -555,7 +555,7 @@ def pipeline_that_has_a_component_with_mutable_input():
             input_list.append("extra_item")
             return {"mangled_list": input_list}
 
-    pipe = Pipeline()
+    pipe = Pipeline(max_runs_per_component=1)
     pipe.add_component("mangler1", InputMangler())
     pipe.add_component("mangler2", InputMangler())
     pipe.add_component("concat1", StringListJoiner())
@@ -607,7 +607,7 @@ def pipeline_that_has_a_component_with_mutable_output_sent_to_multiple_inputs():
     mm1 = MessageMerger()
     mm2 = MessageMerger()
 
-    pipe = Pipeline()
+    pipe = Pipeline(max_runs_per_component=1)
     pipe.add_component("prompt_builder", prompt_builder)
     pipe.add_component("llm", llm)
     pipe.add_component("mm1", mm1)
@@ -663,7 +663,7 @@ def pipeline_that_has_a_greedy_and_variadic_component_after_a_component_with_def
     document_store = InMemoryDocumentStore()
     document_store.write_documents([Document(content="This is a simple document")])
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     template = "Given this documents: {{ documents|join(', ', attribute='content') }} Answer this question: {{ query }}"
     pipeline.add_component("retriever", InMemoryBM25Retriever(document_store=document_store))
     pipeline.add_component("prompt_builder", PromptBuilder(template=template))
@@ -734,7 +734,7 @@ def pipeline_that_has_components_added_in_a_different_order_from_the_order_of_ex
         "Question: {{ query }}"
     )
 
-    pipe = Pipeline()
+    pipe = Pipeline(max_runs_per_component=1)
 
     # The order of this addition is important for the test
     # Do not edit them.
@@ -791,7 +791,7 @@ def pipeline_that_has_a_component_with_only_default_inputs():
         "Question: {{ query }}"
     )
 
-    pipe = Pipeline()
+    pipe = Pipeline(max_runs_per_component=1)
 
     pipe.add_component("retriever", InMemoryBM25Retriever(document_store=doc_store))
     pipe.add_component("prompt_builder", PromptBuilder(template=template))
@@ -889,7 +889,7 @@ def pipeline_that_has_a_component_with_only_default_inputs_as_first_to_run():
         ]
     )
 
-    pipe = Pipeline()
+    pipe = Pipeline(max_runs_per_component=1)
 
     pipe.add_component("prompt_builder", PromptBuilder(template=template))
     pipe.add_component("generator", FakeGenerator())
@@ -968,7 +968,7 @@ def pipeline_that_has_a_component_that_sends_one_of_its_outputs_to_itself():
     target_fixture="pipeline_data",
 )
 def pipeline_that_has_multiple_branches_that_merge_into_a_component_with_a_single_variadic_input():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("add_one", AddFixedValue())
     pipeline.add_component("parity", Remainder(divisor=2))
     pipeline.add_component("add_ten", AddFixedValue(add=10))
@@ -1009,7 +1009,7 @@ def pipeline_that_has_multiple_branches_that_merge_into_a_component_with_a_singl
     target_fixture="pipeline_data",
 )
 def pipeline_that_has_multiple_branches_of_different_lengths_that_merge_into_a_component_with_a_single_variadic_input():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_addition", AddFixedValue(add=2))
     pipeline.add_component("second_addition", AddFixedValue(add=2))
     pipeline.add_component("third_addition", AddFixedValue(add=2))
@@ -1036,7 +1036,7 @@ def pipeline_that_has_multiple_branches_of_different_lengths_that_merge_into_a_c
 
 @given("a pipeline that is linear and returns intermediate outputs", target_fixture="pipeline_data")
 def pipeline_that_is_linear_and_returns_intermediate_outputs():
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_addition", AddFixedValue(add=2))
     pipeline.add_component("second_addition", AddFixedValue())
     pipeline.add_component("double", Double())
@@ -1143,7 +1143,7 @@ def pipeline_that_is_linear_and_returns_intermediate_outputs_from_multiple_socke
         def run(self, value: int):
             return {"value": value * 2, "original": value}
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_addition", AddFixedValue(add=2))
     pipeline.add_component("second_addition", AddFixedValue())
     pipeline.add_component("double", DoubleWithOriginal())
@@ -1184,7 +1184,7 @@ def pipeline_that_has_a_component_with_default_inputs_that_doesnt_receive_anythi
     ]
     router = ConditionalRouter(routes)
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("router", router)
     pipeline.add_component("pb", PromptBuilder(template="Ok, I know, that's {{language}}"))
     pipeline.connect("router.language_2", "pb.language")
@@ -1260,7 +1260,7 @@ def pipeline_that_has_a_component_with_default_inputs_that_doesnt_receive_anythi
     )
     fallback_llm = FakeGenerator()
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("prompt", prompt)
     pipeline.add_component("llm", llm)
     pipeline.add_component("router", router)
@@ -1353,7 +1353,7 @@ def pipeline_that_has_a_loop_and_a_component_with_default_inputs_that_doesnt_rec
     llm = FakeGenerator()
     validator = FakeOutputValidator()
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("prompt_builder", prompt_builder)
 
     pipeline.add_component("llm", llm)
@@ -1450,7 +1450,7 @@ def pipeline_that_has_multiple_components_with_only_default_inputs_and_are_added
         def run(self, prompt: str, generation_kwargs: Optional[Dict[str, Any]] = None):
             return {"replies": ["This is a reply"], "meta": {"meta_key": "meta_value"}}
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component(name="retriever", instance=FakeRetriever())
     pipeline.add_component(name="ranker", instance=FakeRanker())
     pipeline.add_component(name="prompt_builder2", instance=prompt_builder2)
@@ -1594,7 +1594,7 @@ def that_has_a_variadic_component_that_receives_partial_inputs():
                 return {"documents": [Document(id=self._content, content=self._content)]}
             return {"noop": None}
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("first_creator", ConditionalDocumentCreator(content="First document"))
     pipeline.add_component("second_creator", ConditionalDocumentCreator(content="Second document"))
     pipeline.add_component("third_creator", ConditionalDocumentCreator(content="Third document"))
@@ -1641,7 +1641,7 @@ def that_has_a_variadic_component_that_receives_partial_inputs():
 def that_has_an_answer_joiner_variadic_component():
     query = "What's Natural Language Processing?"
 
-    pipeline = Pipeline()
+    pipeline = Pipeline(max_runs_per_component=1)
     pipeline.add_component("answer_builder_1", AnswerBuilder())
     pipeline.add_component("answer_builder_2", AnswerBuilder())
     pipeline.add_component("answer_joiner", AnswerJoiner())
diff --git a/test/core/pipeline/test_pipeline.py b/test/core/pipeline/test_pipeline.py
index e33d4580..9e8c46ef 100644
--- a/test/core/pipeline/test_pipeline.py
+++ b/test/core/pipeline/test_pipeline.py
@@ -11,7 +11,7 @@ from haystack import Document
 from haystack.components.builders import PromptBuilder, AnswerBuilder
 from haystack.components.joiners import BranchJoiner
 from haystack.core.component import component
-from haystack.core.component.types import InputSocket, OutputSocket, Variadic
+from haystack.core.component.types import InputSocket, OutputSocket, Variadic, GreedyVariadic
 from haystack.core.errors import DeserializationError, PipelineConnectError, PipelineDrawingError, PipelineError
 from haystack.core.pipeline import Pipeline, PredefinedPipeline
 from haystack.core.pipeline.base import (
@@ -20,6 +20,7 @@ from haystack.core.pipeline.base import (
     _dequeue_component,
     _enqueue_waiting_component,
     _dequeue_waiting_component,
+    _is_lazy_variadic,
 )
 from haystack.core.serialization import DeserializationCallbacks
 from haystack.testing.factory import component_class
@@ -1628,3 +1629,19 @@ class TestPipeline:
         waiting_queue = [("document_builder", document_builder), ("document_joiner", document_joiner)]
         _dequeue_waiting_component(("document_builder", document_builder), waiting_queue)
         assert waiting_queue == [("document_joiner", document_joiner)]
+
+    def test__is_lazy_variadic(self):
+        VariadicAndGreedyVariadic = component_class(
+            "VariadicAndGreedyVariadic", input_types={"variadic": Variadic[int], "greedy_variadic": GreedyVariadic[int]}
+        )
+        NonVariadic = component_class("NonVariadic", input_types={"value": int})
+        VariadicNonGreedyVariadic = component_class(
+            "VariadicNonGreedyVariadic", input_types={"variadic": Variadic[int]}
+        )
+        NonVariadicAndGreedyVariadic = component_class(
+            "NonVariadicAndGreedyVariadic", input_types={"greedy_variadic": GreedyVariadic[int]}
+        )
+        assert not _is_lazy_variadic(VariadicAndGreedyVariadic())
+        assert not _is_lazy_variadic(NonVariadic())
+        assert _is_lazy_variadic(VariadicNonGreedyVariadic())
+        assert not _is_lazy_variadic(NonVariadicAndGreedyVariadic())