Repository: deepset-ai/haystack
Pull request: feat: Add variable to specify inputs as optional to ConditionalRouter
## Why:

Enhances the ConditionalRouter component by adding support for optional parameters, enabling default/fallback routing behavior when certain inputs are not provided at runtime.

- fixes https://github.com/deepset-ai/haystack/issues/8556

## What:
Introduces `optional_variables` parameter to specify which route variables can be omitted
Improves documentation with clear examples of fallback routing patterns
Implements proper serialization/deserialization for the new field

## How can it be used:
Define routes with fallback behavior when certain parameters are missing:

```python
from haystack import Pipeline
from haystack.components.routers import ConditionalRouter

routes = [
    {
        "condition": '{{ path == "rag" }}',
        "output": "{{ question }}",
        "output_name": "rag_route",
        "output_type": str
    },
    {
        "condition": "{{ True }}",  # fallback route
        "output": "{{ question }}",
        "output_name": "default_route",
        "output_type": str
    }
]

router = ConditionalRouter(routes, optional_variables=["path"])
pipe = Pipeline()
pipe.add_component("router", router)

# When 'path' is provided in the pipeline:
result = pipe.run(data={"router": {"question": "What?", "path": "rag"}})
assert result["router"] == {"rag_route": "What?"}

# When 'path' is not provided, fallback route is taken:
result = pipe.run(data={"router": {"question": "What?"}})
assert result["router"] == {"default_route": "What?"}
```

## How did you test it:
Unit tests verify both direct component usage and pipeline integration
Tests cover simple and complex scenarios with and without optional parameters


Problem statement:
Add variable to specify inputs as optional to ConditionalRouter
**Is your feature request related to a problem? Please describe.**
We are using a conditional router to route to different pipeline branches based on a parameter we insert during runtime. We would like to also set a default branch in case the respective parameter (path) was not inserted at runtime.

Through testing we figured out this works if you run the component individually. So
```python
from haystack.components.routers import ConditionalRouter

routes = [
    {
        "condition": '{{path == "rag"}}',
        "output": "{{question}}",
        "output_name": "normal",
        "output_type": str
    },
    {
        "condition": '{{path == "followup_short"}}',
        "output": "{{question}}",
        "output_name": "followup_short",
        "output_type": str
    },
    {
        "condition": '{{path == "followup_elaborate"}}',
        "output": "{{question}}",
        "output_name": "followup_elaborate",
        "output_type": str
    },
    {
        "condition": "{{ True }}",
        "output": "{{ question }}",
        "output_name": "fallback",
        "output_type": str
    }
]

router = ConditionalRouter(routes)

res = router.run(question="What?")
print(res)
```

However, when inserting it into a Pipeline it no longer works

```python
from haystack import Pipeline
from haystack.components.routers import ConditionalRouter

routes = [
    {
        "condition": '{{path == "rag"}}',
        "output": "{{question}}",
        "output_name": "normal",
        "output_type": str
    },
    {
        "condition": '{{path == "followup_short"}}',
        "output": "{{question}}",
        "output_name": "followup_short",
        "output_type": str
    },
    {
        "condition": '{{path == "followup_elaborate"}}',
        "output": "{{question}}",
        "output_name": "followup_elaborate",
        "output_type": str
    },
    {
        "condition": "{{ True }}",
        "output": "{{ question }}",
        "output_name": "fallback",
        "output_type": str
    }
]

router = ConditionalRouter(routes)
pipe = Pipeline()
pipe.add_component("router", router)


res = pipe.run(data={"router": {"question": "What?"}})
print(res)
```

It would be great to add a way to make parameters (or some of them optional). Perhaps following the PromptBuilder we could add a `required_variables` parameter or do the inverse (to not cause a breaking change) and make `optional_variables` list so that all parameters are required by default (current behavior) and then explicitly add some as optional.

Hints from issue discussion:

Commit message: Add pylint disable=too-many-positional-arguments

Patch to review:
diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py
index 27312f43a1..020fdc3e9b 100644
--- a/haystack/components/routers/conditional_router.py
+++ b/haystack/components/routers/conditional_router.py
@@ -107,12 +107,13 @@ class ConditionalRouter:
     ```
     """
 
-    def __init__(
+    def __init__(  # pylint: disable=too-many-positional-arguments
         self,
         routes: List[Dict],
         custom_filters: Optional[Dict[str, Callable]] = None,
         unsafe: bool = False,
         validate_output_type: bool = False,
+        optional_variables: Optional[List[str]] = None,
     ):
         """
         Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.
@@ -136,11 +137,49 @@ def __init__(
         :param validate_output_type:
             Enable validation of routes' output.
             If a route output doesn't match the declared type a ValueError is raised running.
+        :param optional_variables:
+            A list of variable names that are optional in your route conditions and outputs.
+            If these variables are not provided at runtime, they will be set to `None`.
+            This allows you to write routes that can handle missing inputs gracefully without raising errors.
+
+            Example usage with a default fallback route:
+            ```python
+            routes = [
+                {
+                    "condition": '{{ path == "rag" }}',
+                    "output": "{{ question }}",
+                    "output_name": "rag_route",
+                    "output_type": str
+                },
+                {
+                    "condition": "{{ True }}",  # fallback route
+                    "output": "{{ question }}",
+                    "output_name": "default_route",
+                    "output_type": str
+                }
+            ]
+
+            router = ConditionalRouter(routes, optional_variables=["path"])
+
+            # When 'path' is provided, specific route is taken:
+            result = router.run(question="What?", path="rag")
+            assert result == {"rag_route": "What?"}
+
+            # When 'path' is not provided, fallback route is taken:
+            result = router.run(question="What?")
+            assert result == {"default_route": "What?"}
+            ```
+
+            This pattern is particularly useful when:
+            - You want to provide default/fallback behavior when certain inputs are missing
+            - Some variables are only needed for specific routing conditions
+            - You're building flexible pipelines where not all inputs are guaranteed to be present
         """
         self.routes: List[dict] = routes
         self.custom_filters = custom_filters or {}
         self._unsafe = unsafe
         self._validate_output_type = validate_output_type
+        self.optional_variables = optional_variables or []
 
         # Create a Jinja environment to inspect variables in the condition templates
         if self._unsafe:
@@ -166,7 +205,26 @@ def __init__(
             # extract outputs
             output_types.update({route["output_name"]: route["output_type"]})
 
-        component.set_input_types(self, **{var: Any for var in input_types})
+        # remove optional variables from mandatory input types
+        mandatory_input_types = input_types - set(self.optional_variables)
+
+        # warn about unused optional variables
+        unused_optional_vars = set(self.optional_variables) - input_types if self.optional_variables else None
+        if unused_optional_vars:
+            warn(
+                f"The following optional variables are specified but not used in any route: {unused_optional_vars}. "
+                "Check if there's a typo in variable names.",
+                UserWarning,
+            )
+
+        # add mandatory input types
+        component.set_input_types(self, **{var: Any for var in mandatory_input_types})
+
+        # now add optional input types
+        for optional_var_name in self.optional_variables:
+            component.set_input_type(self, name=optional_var_name, type=Any, default=None)
+
+        # set output types
         component.set_output_types(self, **output_types)
 
     def to_dict(self) -> Dict[str, Any]:
@@ -186,6 +244,7 @@ def to_dict(self) -> Dict[str, Any]:
             custom_filters=se_filters,
             unsafe=self._unsafe,
             validate_output_type=self._validate_output_type,
+            optional_variables=self.optional_variables,
         )
 
     @classmethod
diff --git a/releasenotes/notes/conditional-router-optional-parameters-f02c598d7c751868.yaml b/releasenotes/notes/conditional-router-optional-parameters-f02c598d7c751868.yaml
new file mode 100644
index 0000000000..99056a24da
--- /dev/null
+++ b/releasenotes/notes/conditional-router-optional-parameters-f02c598d7c751868.yaml
@@ -0,0 +1,4 @@
+---
+enhancements:
+  - |
+    Introduces optional parameters in the ConditionalRouter component, enabling default/fallback routing behavior when certain inputs are not provided at runtime. This enhancement allows for more flexible pipeline configurations with graceful handling of missing parameters.
diff --git a/test/components/routers/test_conditional_router.py b/test/components/routers/test_conditional_router.py
index 8ea3f86d92..e0f3552319 100644
--- a/test/components/routers/test_conditional_router.py
+++ b/test/components/routers/test_conditional_router.py
@@ -7,6 +7,7 @@
 
 import pytest
 
+from haystack import Pipeline
 from haystack.components.routers import ConditionalRouter
 from haystack.components.routers.conditional_router import NoRouteSelectedException
 from haystack.dataclasses import ChatMessage
@@ -397,3 +398,147 @@ def test_validate_output_type_with_unsafe(self):
         streams = ["1", "2", "3", "4"]
         with pytest.raises(ValueError, match="Route 'streams' type doesn't match expected type"):
             router.run(streams=streams, message=message)
+
+    def test_router_with_optional_parameters(self):
+        """
+        Test that the router works with optional parameters, particularly testing the default/fallback route
+        when an expected parameter is not provided.
+        """
+        routes = [
+            {"condition": '{{path == "rag"}}', "output": "{{question}}", "output_name": "normal", "output_type": str},
+            {
+                "condition": '{{path == "followup_short"}}',
+                "output": "{{question}}",
+                "output_name": "followup_short",
+                "output_type": str,
+            },
+            {
+                "condition": '{{path == "followup_elaborate"}}',
+                "output": "{{question}}",
+                "output_name": "followup_elaborate",
+                "output_type": str,
+            },
+            {"condition": "{{ True }}", "output": "{{ question }}", "output_name": "fallback", "output_type": str},
+        ]
+
+        router = ConditionalRouter(routes, optional_variables=["path"])
+
+        # Test direct component usage
+        result = router.run(question="What?")
+        assert result == {"fallback": "What?"}, "Default route should be taken when 'path' is not provided"
+
+        # Test with path parameter
+        result = router.run(question="What?", path="rag")
+        assert result == {"normal": "What?"}, "Specific route should be taken when 'path' is provided"
+
+        pipe = Pipeline()
+        pipe.add_component("router", router)
+
+        # Test pipeline without path parameter
+        result = pipe.run(data={"router": {"question": "What?"}})
+        assert result["router"] == {
+            "fallback": "What?"
+        }, "Default route should work in pipeline when 'path' is not provided"
+
+        # Test pipeline with path parameter
+        result = pipe.run(data={"router": {"question": "What?", "path": "followup_short"}})
+        assert result["router"] == {"followup_short": "What?"}, "Specific route should work in pipeline"
+
+    def test_router_with_multiple_optional_parameters(self):
+        """
+        Test ConditionalRouter with a mix of mandatory and optional parameters,
+        exploring various combinations of provided/missing optional variables.
+        """
+        routes = [
+            {
+                "condition": '{{mode == "chat" and language == "en" and source == "doc"}}',
+                "output": "{{question}}",
+                "output_name": "en_doc_chat",
+                "output_type": str,
+            },
+            {
+                "condition": '{{mode == "qa" and source == "web"}}',
+                "output": "{{question}}",
+                "output_name": "web_qa",
+                "output_type": str,
+            },
+            {
+                "condition": '{{mode == "qa" and source == "doc"}}',
+                "output": "{{question}}",
+                "output_name": "doc_qa",
+                "output_type": str,
+            },
+            {
+                "condition": '{{mode == "chat" and language == "en"}}',
+                "output": "{{question}}",
+                "output_name": "en_chat",
+                "output_type": str,
+            },
+            {
+                "condition": '{{mode == "chat"}}',  # fallback for chat without language
+                "output": "{{question}}",
+                "output_name": "default_chat",
+                "output_type": str,
+            },
+            {
+                "condition": "{{ True }}",  # global fallback
+                "output": "{{question}}",
+                "output_name": "fallback",
+                "output_type": str,
+            },
+        ]
+
+        # There are four variables in the routes:
+        # - mandatory: mode, question (always must be provided) or we'll route to fallback
+        # - optional: source, language
+        router = ConditionalRouter(routes, optional_variables=["source", "language"])
+
+        # Test with mandatory parameter only
+        result = router.run(question="What?", mode="chat")
+        assert result == {"default_chat": "What?"}, "Should use chat fallback when language not provided"
+
+        # Test with all parameters provided
+        result = router.run(question="What?", mode="chat", language="en", source="doc")
+        assert result == {"en_doc_chat": "What?"}, "Should use specific route when all params provided"
+
+        # Test with different mandatory value and one optional
+        result = router.run(question="What?", mode="qa", source="web")
+        assert result == {"web_qa": "What?"}, "Should route qa with source correctly"
+
+        # Test with mandatory the routes to fallback
+        result = router.run(question="What?", mode="qa")
+        assert result == {"fallback": "What?"}, "Should use global fallback for qa without source"
+
+        # Test in pipeline
+        pipe = Pipeline()
+        pipe.add_component("router", router)
+
+        # Test pipeline with mandatory only
+        result = pipe.run(data={"router": {"question": "What?", "mode": "chat"}})
+        assert result["router"] == {"default_chat": "What?"}, "Pipeline should handle missing optionals"
+
+        # Test pipeline with mandatory and one optional
+        result = pipe.run(data={"router": {"question": "What?", "mode": "qa", "source": "doc"}})
+        assert result["router"] == {"doc_qa": "What?"}, "Pipeline should handle all parameters"
+
+        # Test pipeline with mandatory and both optionals
+        result = pipe.run(data={"router": {"question": "What?", "mode": "chat", "language": "en", "source": "doc"}})
+        assert result["router"] == {"en_doc_chat": "What?"}, "Pipeline should handle all parameters"
+
+    def test_warns_on_unused_optional_variables(self):
+        """
+        Test that a warning is raised when optional_variables contains variables
+        that are not used in any route conditions or outputs.
+        """
+        routes = [
+            {"condition": '{{mode == "chat"}}', "output": "{{question}}", "output_name": "chat", "output_type": str},
+            {"condition": "{{ True }}", "output": "{{question}}", "output_name": "fallback", "output_type": str},
+        ]
+
+        # Initialize with unused optional variables and capture warning
+        with pytest.warns(UserWarning, match="optional variables"):
+            router = ConditionalRouter(routes=routes, optional_variables=["unused_var1", "unused_var2"])
+
+        # Verify router still works normally
+        result = router.run(question="What?", mode="chat")
+        assert result == {"chat": "What?"}


Code review comments:
diff_hunk: @@ -136,11 +137,49 @@ def __init__(
         :param validate_output_type:
             Enable validation of routes' output.
             If a route output doesn't match the declared type a ValueError is raised running.
+        :param optional_variables:
+            A list of variable names that are optional in your route conditions and outputs.
+            If these variables are not provided at runtime, they will be set to `None`.
+            This allows you to write routes that can handle missing inputs gracefully without raising errors.
+
+            Example usage with a default fallback route:
+            ```python
+            routes = [
+                {
+                    "condition": '{{ path == "rag" }}',
+                    "output": "{{ question }}",
+                    "output_name": "rag_route",
+                    "output_type": str
+                },
+                {
+                    "condition": "{{ True }}",  # fallback route
+                    "output": "{{ question }}",
+                    "output_name": "default_route",
+                    "output_type": str
+                }
+            ]
+
+            router = ConditionalRouter(routes, optional_variables=["path"])
+
+            # When 'path' is provided, specific route is taken:
+            result = router.run(question="What?", path="rag")
+            assert result == {"rag_route": "What?"}
+
+            # When 'path' is not provided, fallback route is taken:
+            result = router.run(question="What?")
+            assert result == {"default_route": "What?"}
path: haystack/components/routers/conditional_router.py
text: @user1:
Good example! Except this example actually works pre your changes. To get the error we were originally seeing you would need to put it in a Pipeline first. Even a single component pipeline would do. Otherwise it's a bit misleading here since this example would actually work without specifying the `optional_variables` parameter.