Custom components: Adding warning when not all inputs can be obtained

[CharlesCousyn]

Is your feature request related to a problem? Please describe. Here a simplified RAG pipeline (where a directly inject documents for testing): The purpose of the latter is to offer a way of bypassing the generator when no documents is available using a ConditionalRouter.

The very last component of the pipeline is a custom component whose aim is to aggregate all variables in a payload that will be provided to build an API . Its code is as follows:

@component
class PayloadBuilder2:
    """
    A component to build the payload
    """

    @component.output_types(payload=Dict)
    def run(
            self,
            documents: List[Document],
            replies: List[ChatMessage],
            no_documents: str,
    ):
        print("PayloadBuilder.run")
        return {
            "payload": {
                "documents": documents,
                "no_documents": no_documents,
                "replies": replies,
            }
        }

The problem is that no matter what we send, the pipeline will return an empty dictionary:

{}

By debugging, i found that the the reason is that my custom component necessarily expects to have a value for no_documents input in order to be allowed to run. The fix in my case was the following: Replace:

   no_documents: str,

By

   no_documents: str="",

The default parameter allow the component to run even if the pipeline don't go through this no_documents branch!

Describe the solution you'd like Could it be possible to get a warning/message at runtime or creation of the pipeline when a component that should be run cannot because it will never have its requested value? ( because of a CondionalRouter for example)

Describe alternatives you've considered A clear and concise description of any alternative solutions or features you've considered.

Additional context Add any other context or screenshots about the feature request here.

[sjrl]

@julian-risch This still looks to be an issue. Here is some example code

import logging

logging.basicConfig(level=logging.DEBUG)

from typing import Dict, List

from haystack import Pipeline, component
from haystack.components.routers.conditional_router import ConditionalRouter

c = ConditionalRouter(
    [
        {
            "condition": "{{streams|length < 2}}",
            "output": "{{query}}",
            "output_type": str,
            "output_name": "query"
        },
        {
            "condition": "{{streams|length >= 2}}",
            "output": "{{streams}}",
            "output_type": List[int],
            "output_name": "streams",
        },
    ]
)


@component
class PayloadBuilder2:
    """
    A component to build the payload
    """

    @component.output_types(payload=Dict)
    def run(
        self,
        streams: List[int],
        query: str,
    ):
        print("PayloadBuilder.run")
        return {
            "payload": {
                "streams": streams,
                "query": query,
            }
        }


pipe = Pipeline()
pipe.add_component("router", c)
pipe.add_component("payload_builder", PayloadBuilder2())

pipe.connect("router.streams", "payload_builder.streams")
pipe.connect("router.query", "payload_builder.query")

res = pipe.run(data={"router": {"streams": [1, 2, 3], "query": "Haystack"}})
# returns {}