Repository: deepset-ai/haystack
Pull request: feat: Add support for multiple outputs in ConditionalRouter
### Related Issues

- fixes #9238 

### Proposed Changes:
- updated output field valdation and input to allow multiple values

### How did you test it?

- added unit tests
- fixed breaking tests in test_conditional_router.py

### Notes for the reviewer

<!-- E.g. point out section where the reviewer  -->

### Checklist

- I have read the [contributors guidelines](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md) and the [code of conduct](https://github.com/deepset-ai/haystack/blob/main/code_of_conduct.txt)
- I have updated the related issue with new insights and changes
- I added unit tests and updated the docstrings
- I've used one of the [conventional commit types](https://www.conventionalcommits.org/en/v1.0.0/) for my PR title: `fix:`, `feat:`, `build:`, `chore:`, `ci:`, `docs:`, `style:`, `refactor:`, `perf:`, `test:` and added `!` in case the PR includes breaking changes.
- I documented my code
- I ran [pre-commit hooks](https://github.com/deepset-ai/haystack/blob/main/CONTRIBUTING.md#installation) and fixed any issue




Problem statement:
Allow `ConditionalRouter` to have multiple outputs per condition
**Is your feature request related to a problem? Please describe.**
ConditionalRouter is very useful and quite powerful for switching between branches in a pipeline. However, there are times where I would like to route multiple things through one connection. This makes it easier to see the flow of the pipeline by funnelling the connections through a choke point. 

Additionally, it can help prevent triggering uneccessary component executions by making sure all inputs are only passed on conditionally.

**Describe the solution you'd like**
The solution I'd like is something like 
```python
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}}", "{{query}}"],  # <-- CHANGE: Make this a List of str
            "output_type": [List[int], str],  # CHANGE: Make this a List of Types
            "output_name": "streams",
        },
    ]
)
```
and then we can set multiple outputs and output types for a single route.

**Additional context**
A work around to this that gets clunk very quickly is to:
1) Return a dict as output from the Router
2) Have as many OutputAdapters as there are keys in the dict to convert them into single connections. So if there are 4 keys in the dict you'd need 4 OutputAdapters to access them and forward them to the next component. 

cc @ju-gu @tstadel

Hints from issue discussion:

Commit message: feat: Add support for multiple outputs in ConditionalRouter

Patch to review:
diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py
index 8f94882b26..59fc36a6f2 100644
--- a/haystack/components/routers/conditional_router.py
+++ b/haystack/components/routers/conditional_router.py
@@ -203,11 +203,19 @@ def __init__(  # pylint: disable=too-many-positional-arguments
 
         for route in routes:
             # extract inputs
-            route_input_names = self._extract_variables(self._env, [route["output"], route["condition"]])
+            route_input_names = self._extract_variables(
+                self._env,
+                [route["condition"]] + (route["output"] if isinstance(route["output"], list) else [route["output"]]),
+            )
             input_types.update(route_input_names)
 
             # extract outputs
-            output_types.update({route["output_name"]: route["output_type"]})
+            output_names = route["output_name"] if isinstance(route["output_name"], list) else [route["output_name"]]
+            output_types_list = (
+                route["output_type"] if isinstance(route["output_type"], list) else [route["output_type"]]
+            )
+
+            output_types.update(dict(zip(output_names, output_types_list)))
 
         # remove optional variables from mandatory input types
         mandatory_input_types = input_types - set(self.optional_variables)
@@ -306,27 +314,45 @@ def run(self, **kwargs):
                     rendered = ast.literal_eval(rendered)
                 if not rendered:
                     continue
-                # We now evaluate the `output` expression to determine the route output
-                t_output = self._env.from_string(route["output"])
-                output = t_output.render(**kwargs)
-                # We suppress the exception in case the output is already a string, otherwise
-                # we try to evaluate it and would fail.
-                # This must be done cause the output could be different literal structures.
-                # This doesn't support any user types.
-                with contextlib.suppress(Exception):
-                    if not self._unsafe:
-                        output = ast.literal_eval(output)
+
+                # Handle multiple outputs
+                outputs = route["output"] if isinstance(route["output"], list) else [route["output"]]
+                output_types = (
+                    route["output_type"] if isinstance(route["output_type"], list) else [route["output_type"]]
+                )
+                output_names = (
+                    route["output_name"] if isinstance(route["output_name"], list) else [route["output_name"]]
+                )
+
+                result = {}
+                for output, output_type, output_name in zip(outputs, output_types, output_names):
+                    # Evaluate output template
+                    t_output = self._env.from_string(output)
+                    output_value = t_output.render(**kwargs)
+
+                    # We suppress the exception in case the output is already a string, otherwise
+                    # we try to evaluate it and would fail.
+                    # This must be done cause the output could be different literal structures.
+                    # This doesn't support any user types.
+                    with contextlib.suppress(Exception):
+                        if not self._unsafe:
+                            output_value = ast.literal_eval(output_value)
+
+                    # Validate output type if needed
+                    if self._validate_output_type and not self._output_matches_type(output_value, output_type):
+                        raise ValueError(f"Route '{output_name}' type doesn't match expected type")
+
+                    result[output_name] = output_value
+
+                return result
+
             except Exception as e:
+                # If this was a type‐validation failure, let it propagate as a ValueError
+                if isinstance(e, ValueError):
+                    raise
                 msg = f"Error evaluating condition for route '{route}': {e}"
                 raise RouteConditionException(msg) from e
 
-            if self._validate_output_type and not self._output_matches_type(output, route["output_type"]):
-                msg = f"""Route '{route["output_name"]}' type doesn't match expected type"""
-                raise ValueError(msg)
-
-            # and return the output as a dictionary under the output_name key
-            return {route["output_name"]: output}
-
         raise NoRouteSelectedException(f"No route fired. Routes: {self.routes}")
 
     def _validate_routes(self, routes: List[Dict]):
@@ -347,9 +373,23 @@ def _validate_routes(self, routes: List[Dict]):
                 raise ValueError(
                     f"Route must contain 'condition', 'output', 'output_type' and 'output_name' fields: {route}"
                 )
-            for field in ["condition", "output"]:
-                if not self._validate_template(self._env, route[field]):
-                    raise ValueError(f"Invalid template for field '{field}': {route[field]}")
+
+            # Validate outputs are consistent
+            outputs = route["output"] if isinstance(route["output"], list) else [route["output"]]
+            output_types = route["output_type"] if isinstance(route["output_type"], list) else [route["output_type"]]
+            output_names = route["output_name"] if isinstance(route["output_name"], list) else [route["output_name"]]
+
+            # Check lengths match
+            if not (len(outputs) == len(output_types) == len(output_names)):
+                raise ValueError(f"Route output, output_type and output_name must have same length: {route}")
+
+            # Validate templates
+            if not self._validate_template(self._env, route["condition"]):
+                raise ValueError(f"Invalid template for condition: {route['condition']}")
+
+            for output in outputs:
+                if not self._validate_template(self._env, output):
+                    raise ValueError(f"Invalid template for output: {output}")
 
     def _extract_variables(self, env: Environment, templates: List[str]) -> Set[str]:
         """
diff --git a/releasenotes/notes/multiple-outputs-conditional-router-c2f0caad3d3f8ce5.yaml b/releasenotes/notes/multiple-outputs-conditional-router-c2f0caad3d3f8ce5.yaml
new file mode 100644
index 0000000000..424a359ceb
--- /dev/null
+++ b/releasenotes/notes/multiple-outputs-conditional-router-c2f0caad3d3f8ce5.yaml
@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Add support for multiple outputs in ConditionalRouter
diff --git a/test/components/routers/test_conditional_router.py b/test/components/routers/test_conditional_router.py
index 0a86613454..4d57578999 100644
--- a/test/components/routers/test_conditional_router.py
+++ b/test/components/routers/test_conditional_router.py
@@ -574,3 +574,34 @@ def test_router_to_dict_does_not_mutate_routes(self):
         assert new_router.routes == router.routes
         assert new_router.routes[0]["output_type"] is str
         assert new_router.routes[0]["output_type"] is original_output_type
+
+    def test_multiple_outputs_per_route(self):
+        """Test that router handles multiple outputs per route correctly"""
+        routes = [
+            {
+                "condition": "{{streams|length >= 2}}",
+                "output": ["{{streams}}", "{{query}}"],
+                "output_type": [List[int], str],
+                "output_name": ["streams", "query"],
+            }
+        ]
+        router = ConditionalRouter(routes)
+
+        # Test with valid input
+        result = router.run(streams=[1, 2, 3], query="test")
+        assert result == {"streams": [1, 2, 3], "query": "test"}
+
+    def test_multiple_outputs_validation(self):
+        """Test validation of routes with multiple outputs"""
+        # Test mismatched lengths
+        with pytest.raises(ValueError, match="must have same length"):
+            ConditionalRouter(
+                [
+                    {
+                        "condition": "{{streams|length >= 2}}",
+                        "output": ["{{streams}}", "{{query}}"],
+                        "output_type": [List[int]],
+                        "output_name": ["streams"],
+                    }
+                ]
+            )


Code review comments:
diff_hunk: @@ -347,9 +373,23 @@ def _validate_routes(self, routes: List[Dict]):
                 raise ValueError(
                     f"Route must contain 'condition', 'output', 'output_type' and 'output_name' fields: {route}"
                 )
-            for field in ["condition", "output"]:
-                if not self._validate_template(self._env, route[field]):
-                    raise ValueError(f"Invalid template for field '{field}': {route[field]}")
+
+            # Validate outputs are consistent
+            outputs = route["output"] if isinstance(route["output"], list) else [route["output"]]
+            output_types = route["output_type"] if isinstance(route["output_type"], list) else [route["output_type"]]
+            output_names = route["output_name"] if isinstance(route["output_name"], list) else [route["output_name"]]
+
+            # Check lengths match
+            if not (len(outputs) == len(output_types) == len(output_names)):
path: haystack/components/routers/conditional_router.py
text: @user1:
```suggestion
            if not len(outputs) == len(output_types) == len(output_names):
```
remove extra parens to make pylint happy