Repository: deepset-ai/haystack
Pull request: feat: Add `ConditionalRouter` Haystack 2.x component
### Why:
- Enable generic and conditionally expressive pipeline routing functionality by introducing a new `Router` component.
- The `Router` component orchestrates the flow of data by evaluating specified route conditions to determine the appropriate route among a set of provided route alternatives.
- fixes https://github.com/deepset-ai/haystack/issues/6109

### What:
- Added a new `Router` class to `haystack/preview/components/routers`.
- Updated `__init__.py` to include the new `Router` class.

### How can it be used:
- Import and utilize the `Router` component to manage and route connections in your pipelines.
- Here is an example:

 In this example, we create a `Router` component with two routes. The first route will be selected if the number of streams is less than 2, and will output the `query` variable. The second route will be selected if the number of streams is 2 or more, and will output the `streams` variable. We also specify the routing variables, which are `query` and `streams`. These variables need to be provided in the pipeline `run()` method. Routing variables can be used in route conditions and as route output values. 
  ```python

routes = [
    {"condition": "len(streams) < 2", "output": "query", "output_type": str},
    {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]}
]

router = Router(routes=routes, routing_variables=["query", "streams"])


# the second route from above should be selected
kwargs = {"streams": [1, 2, 3], "query": "test"}
result = router.run(**kwargs)
assert result == {"streams": [1, 2, 3]}

# the first route from above should be selected
kwargs = {"streams": [1], "query": "test"}
result = router.run(**kwargs)
assert result == {"query": "test"}
  ```    

### How did you test it:
- Unit tests were added to ensure that the new `Router` component works correctly on the component level. A real-world example is available in this [colab](https://colab.research.google.com/drive/1sg-Q6QKN0dzUgIj6VY19kkC5o4CqtzXf?usp=sharing)

### Notes for reviewer:
- This is not a final version but more of a start of a conversation in the direction of expressive conditional routing in Haystack 2.x. DO NOT INTEGRATE

Problem statement:
Add Conditional Routing in Haystack 2.x Pipelines
### Description:

In the current state of Haystack 2.x pipelines, we can create pipelines by connecting output slots to input slots:

```python
pipe.connect("fetcher.streams", "converter.sources")
pipe.connect("converter.documents", "text_splitter.documents")
```

However, there isn't direct support for conditional routing based on user-specified boolean conditions. For example, when `LinkContentFetcher` returns too few streams (i.e. perhaps it got blocked) we want to make decisions based on the number of streams received. If we receive too few streams, we might want to reformulate the query in some `QueryRewriter` component, and then rerun it via  `LinkContentFetcher` etc etc.

### Describe the solution you'd like:

Introduce a well-known Haystack component, possibly named `ConnectionRouter` (or a similar fitting name), which supports registration of boolean expressions as keys and component input slots as values:

```python
fetcher = LinkContentFetcher()
router = ConnectionRouter()
converter = HTMLToDocument()
query_rewriter = QueryRewriter()

router.register(lambda streams: len(streams) < 2, query_rewriter)
router.register(lambda streams: len(streams) >= 2, converter)
...
pipe.add_component("fetcher", fetcher)
pipe.add_component("router", router)
pipe.add_component("query_rewriter", query_rewriter)
pipe.add_component("converter", html_converter)
....
pipe.connect("fetcher.streams", "router.input")
pipe.connect("converter.documents", "text_splitter.documents")
pipe.connect("query_rewriter.urls", "fetcher.urls")
```

By utilizing such a component, users will have the flexibility to implement custom branching logic in their pipelines based on the component result object boolean expressions and not only on output slots.

### Describe alternatives you've considered

The alternative seems to be creating a custom component for each branching scenario (use case)

### Additional context

Having such a generic `ConnectionRouter` would offer immediate benefits:
1. **Flexibility**: Allows custom conditional, super-detailed, custom branching in the pipeline, making Haystack more adaptable to custom user scenarios.
2. **Readability**: Having a dedicated router component can make the pipeline more readable, especially when complex routing decisions are needed.
3. **Usability**: Users can leverage lambda functions or any other callable to define custom conditions, making it powerful yet simple.

Hints from issue discussion:
Another use case where I needed such a `ConnectionRouter` is with LLM results and function calling. In step 4 of the colab [notebook](https://colab.research.google.com/drive/1hPyK7DKXuMRRm03uE7-h-oG0X4h3Lvpx), I want to inspect ChatMessage to see if it's metadata indicates `finish_reason` to be `function_call`. In such a case, I'd like to route ChatMessage to a component A that'll resolve the function, invoke it, and append the response to messages. By routing the output slot of component A back to LLM, we can complete the function call (step 6 of the notebook).  
This looks great and the code example really helped me understanding what we want to achieve here. I have two concerns with the solution, let me illustrate them.

1. The `register` API doesn't play well with serialization - imagine I want to rewrite the example in Yaml format:
```yaml
components:
  converter:
    type: HTMLToDocument

  fetcher:
    init_parameters:
      raise_on_failure: true
      retry_attempts: 2
      timeout: 3
      user_agents:
      - haystack/LinkContentFetcher/1.22.0rc0
    type: LinkContentFetcher

  router:
    type: ConnectionRouter
    # how do I express "register"?

connections:
- receiver: router.input
  sender: fetcher.streams
```

One alternative solution might be leveraging the new filters (see the [proposal](https://github.com/deepset-ai/haystack/blob/main/proposals/text/6001-document-store-filter-rework.md)) so that we can express conditions in text format instead of Python code:

```python

## Before:
# router = ConnectionRouter()
# router.register(lambda streams: len(streams) < 2, query_rewriter)
# router.register(lambda streams: len(streams) >= 2, converter)

# After
routes = {
  query_rewriter: { "input": "streams", "operator": "<", "value": "2" },
  converter: { "input": "streams", "operator": ">=", "value": "2" },
}
router = ConnectionRouter(routes=routes)

# ... rest stays the same
```
which would be easy to express in Yaml:
```yaml
  router:
    type: ConnectionRouter
    init_parameters:
      routes: 
        query_rewriter: { "input": "streams", "operator": "<", "value": "2" }
        converter: { "input": "streams", "operator": ">=", "value": "2" }
# ...
```

2. How do you connect the router to the downstream components? If you draw the pipeline of the example, `query_rewriter` and `converter` would have no inputs.
@masci I like in general the idea of using filters, but I'm not sure they're expressive enough. One may want to route objects (think `Answers`, `ChatMessages`,`ByteStreams`, even just `Documents`) by the content of some attribute. Would the filters manage to represent that?
I'm also a bit perplexed by the fact that in the examples, `router` never appears as the source in any `connect` statement. An oversight? Or is it meant to be that way? I was thinking that the `register()` function would have a signature like:

```python
def register(self, input_name: str, input_type: Type, function: Callable, output_name: str, output_type: Type):
```

while right now seems to be:

```python
def register(self, function: Callable, component_to_connect_to: Component):
```

which I think Canals wouldn't support right now.

Commit message: Add release note

Patch to review:
diff --git a/haystack/preview/components/routers/__init__.py b/haystack/preview/components/routers/__init__.py
index e049d63b87..f24413ddb3 100644
--- a/haystack/preview/components/routers/__init__.py
+++ b/haystack/preview/components/routers/__init__.py
@@ -1,4 +1,6 @@
 from haystack.preview.components.routers.file_type_router import FileTypeRouter
 from haystack.preview.components.routers.metadata_router import MetadataRouter
+from haystack.preview.components.routers.router import Router
 
-__all__ = ["FileTypeRouter", "MetadataRouter"]
+
+__all__ = ["FileTypeRouter", "MetadataRouter", "Router"]
diff --git a/haystack/preview/components/routers/router.py b/haystack/preview/components/routers/router.py
new file mode 100644
index 0000000000..dbe00dca24
--- /dev/null
+++ b/haystack/preview/components/routers/router.py
@@ -0,0 +1,145 @@
+import ast
+import logging
+from typing import List, Dict, Any
+
+from haystack.preview import component
+
+logger = logging.getLogger(__name__)
+
+
+class NoRouteSelectedException(Exception):
+    """Exception raised when no route is selected in Router."""
+
+    def __init__(self, routes, routing_variables, message=None):
+        self.routes = routes
+        self.routing_variables = routing_variables
+
+        super().__init__(
+            (
+                f"No route fired. Routes: {routes}, " f"Routing Variables: {routing_variables}"
+                if message is None
+                else message
+            )
+        )
+
+
+class RouteConditionException(Exception):
+    """Exception raised when there is an error parsing or evaluating the condition expression in Router."""
+
+    def __init__(self, route_directive, message=None):
+        self.route_directive = route_directive
+        super().__init__(f"Error evaluating condition for route: {route_directive}" if message is None else message)
+
+
+class Router:
+    """
+    The Router class orchestrates the flow of data by evaluating specified route conditions
+    to determine the appropriate route among a set of provided route alternatives.
+
+    To use a Router in Haystack 2.x pipelines we first define a list called routes, where each element
+    is a dictionary representing a route.
+
+    Each route dictionary contains three keys: condition, output, and output_type.
+
+    The condition is a string containing a Python expression that will be evaluated to determine if
+    this route should be selected. The output is a string specifying the name of the output slot for
+    this route, and output_type is a string representation of the expected type of the output.
+
+    After specifying routes list we need to also provide routing variables - variables that will be
+    available for use in the condition expressions within the routes as well as the outputs of the router.
+
+    Example:
+
+        In this example, we create a `Router` instance with two routes.
+        The first route will be selected if the number of streams is less than 2,
+        and will output the `query` variable. The second route will be selected
+        if the number of streams is 2 or more, and will output the `streams` variable.
+        We also specify the routing variables, which are `query` and `streams`. These
+        variables need to be provided in the pipeline `run()` method.
+
+        ```python
+
+            routes = [
+                {"condition": "len(streams) < 2", "output": "query", "output_type": "str"},
+                {"condition": "len(streams) >= 2", "output": "streams", "output_type": "List[ByteStream]"}
+            ]
+
+            router = Router(routes=routes, routing_variables=["query", "streams"])
+        ```
+    """
+
+    def __init__(self, routes: List[Dict], routing_variables: List[str]):
+        """
+        Initialize the Router with a list of routes and the routing variables.
+
+        :param routes: A list of dictionaries, each representing a route with a
+                       boolean condition expression (`condition`), an output slot (`output`),
+                       and the output type as a string representation (`output_type`).
+
+        :param routing_variables: A list of additional pipeline variables that are
+                       used in the boolean condition expressions or as outputs of the router.
+                       These variables should be provided by either the pipeline `run()`
+                       method or by a previous component to the router in the pipeline.
+
+        """
+        self._validate_routes(routes)
+        self._validate_routing_variables(routing_variables)
+        self.routes = routes
+        self.routing_variables = routing_variables
+
+        # Compile the condition expressions for each route
+        self.compiled_expressions = []
+        for route_directive in self.routes:
+            expr_ast = ast.parse(route_directive["condition"], mode="eval")
+            try:
+                compiled_expr = compile(expr_ast, filename="<string>", mode="eval")
+            except Exception as e:
+                raise RouteConditionException(route_directive, "Invalid route condition expression.") from e
+            self.compiled_expressions.append(compiled_expr)
+
+        component.set_input_types(self, **{var: Any for var in routing_variables})
+
+        all_output_types = {}
+        for route in routes:
+            all_output_types.update({route["output"]: route["output_type"]})
+        component.set_output_types(self, **all_output_types)
+
+    def run(self, **kwargs):
+        """
+        Executes the routing logic by evaluating the specified boolean condition expressions
+        for each route in the order they are listed. The method directs the flow
+        of data to the output slot specified in the first route whose expression
+        evaluates to True. If no route's expression evaluates to True, an exception
+        is raised.
+
+        :param kwargs: A dictionary containing the pipeline variables, which should
+                   include all variables listed in `routing_variables`.
+
+        :return: A dictionary containing the output slot and the corresponding result,
+             based on the first route whose expression evaluates to True.
+
+        :raises NoRouteSelectedException: If no route's expression evaluates to True.
+        """
+        local_vars = {context_var: kwargs[context_var] for context_var in self.routing_variables}
+        for i, route_directive in enumerate(self.routes):
+            try:
+                result = eval(self.compiled_expressions[i], local_vars)
+            except Exception as e:
+                raise RouteConditionException(route_directive) from e
+            if result:
+                output_slot = route_directive["output"]
+                return {output_slot: kwargs[output_slot]}
+
+        raise NoRouteSelectedException(routes=self.routes, routing_variables=self.routing_variables)
+
+    def _validate_routes(self, routes: List[Dict]):
+        for route in routes:
+            if not isinstance(route, dict):
+                raise ValueError("Each route must be a dictionary.")
+            if "condition" not in route or "output" not in route or "output_type" not in route:
+                raise ValueError("Each route must contain 'condition', 'output', and 'output_type' keys.")
+
+    def _validate_routing_variables(self, routing_variables: List[str]):
+        for var in routing_variables:
+            if not isinstance(var, str):
+                raise ValueError("Each routing variable must be a string.")
diff --git a/releasenotes/notes/add-router-f1f0cec79b1efe9a.yaml b/releasenotes/notes/add-router-f1f0cec79b1efe9a.yaml
new file mode 100644
index 0000000000..3b54c653dc
--- /dev/null
+++ b/releasenotes/notes/add-router-f1f0cec79b1efe9a.yaml
@@ -0,0 +1,6 @@
+---
+preview:
+  - |
+    Introduced a new `Router` component to enhance the conditional pipeline routing capabilities.
+    The Router component orchestrates the flow of data by evaluating specified route conditions
+    to determine the appropriate route among a set of provided route alternatives.
diff --git a/test/preview/components/routers/test_router.py b/test/preview/components/routers/test_router.py
new file mode 100644
index 0000000000..21887ca656
--- /dev/null
+++ b/test/preview/components/routers/test_router.py
@@ -0,0 +1,126 @@
+from typing import List
+
+import pytest
+
+from haystack.preview.components.routers import Router
+from haystack.preview.components.routers.router import NoRouteSelectedException
+
+
+class TestRouter:
+    @pytest.mark.unit
+    def test_router_initialized_with_routes_and_routing_variables(self):
+        """
+        Router can be initialized with a list of routes and routing variables
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", "streams"]
+
+        router = Router(routes=routes, routing_variables=routing_variables)
+
+        assert router.routes == routes
+        assert router.routing_variables == routing_variables
+
+    @pytest.mark.unit
+    def test_router_evaluate_condition_expressions(self):
+        """
+        Router can evaluate the specified boolean condition expressions for each route in the order they are listed
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", "streams"]
+
+        router = Router(routes=routes, routing_variables=routing_variables)
+
+        # first route should be selected
+        kwargs = {"streams": [1, 2, 3], "query": "test"}
+        result = router.run(**kwargs)
+        assert result == {"streams": [1, 2, 3]}
+
+        # second route should be selected
+        kwargs = {"streams": [1], "query": "test"}
+        result = router.run(**kwargs)
+        assert result == {"query": "test"}
+
+    @pytest.mark.unit
+    def test_router_no_route(self):
+        """
+        Router raises an exception if no route's expression evaluates to True
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            {"condition": "len(streams) >= 5", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", "streams"]
+
+        router = Router(routes=routes, routing_variables=routing_variables)
+
+        # should raise an exception
+        kwargs = {"streams": [1, 2, 3], "query": "test"}
+        with pytest.raises(NoRouteSelectedException):
+            router.run(**kwargs)
+
+    @pytest.mark.unit
+    def test_router_direct_flow_to_output_slot(self):
+        """
+        Router can direct the flow of data to the output slot specified in the first route
+        whose expression evaluates to True
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", "streams"]
+
+        router = Router(routes=routes, routing_variables=routing_variables)
+
+        kwargs = {"streams": [1], "query": "test"}
+        result = router.run(**kwargs)
+
+        assert "query" in result
+
+    @pytest.mark.unit
+    def test_router_raises_value_error_if_route_not_dictionary(self):
+        """
+        Router raises a ValueError if each route is not a dictionary
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            ["len(streams) >= 2", "streams", List[int]],
+        ]
+        routing_variables = ["query", "streams"]
+
+        with pytest.raises(ValueError):
+            Router(routes=routes, routing_variables=routing_variables)
+
+    @pytest.mark.unit
+    def test_router_raises_value_error_if_route_missing_keys(self):
+        """
+        Router raises a ValueError if each route does not contain 'condition', 'output', and 'output_type' keys
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query"},
+            {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", "streams"]
+
+        with pytest.raises(ValueError):
+            Router(routes=routes, routing_variables=routing_variables)
+
+    @pytest.mark.unit
+    def test_router_raises_value_error_if_routing_variable_not_string(self):
+        """
+        Router raises a ValueError if each routing variable is not a string
+        """
+        routes = [
+            {"condition": "len(streams) < 2", "output": "query", "output_type": str},
+            {"condition": "len(streams) >= 2", "output": "streams", "output_type": List[int]},
+        ]
+        routing_variables = ["query", 123]
+
+        with pytest.raises(ValueError):
+            Router(routes=routes, routing_variables=routing_variables)


Code review comments:
diff_hunk: @@ -0,0 +1,145 @@
+import ast
+import logging
+from typing import List, Dict, Any
+
+from haystack.preview import component
+
+logger = logging.getLogger(__name__)
+
+
+class NoRouteSelectedException(Exception):
+    """Exception raised when no route is selected in Router."""
+
+    def __init__(self, routes, routing_variables, message=None):
+        self.routes = routes
+        self.routing_variables = routing_variables
+
+        super().__init__(
+            (
+                f"No route fired. Routes: {routes}, " f"Routing Variables: {routing_variables}"
+                if message is None
+                else message
+            )
+        )
+
+
+class RouteConditionException(Exception):
+    """Exception raised when there is an error parsing or evaluating the condition expression in Router."""
+
+    def __init__(self, route_directive, message=None):
+        self.route_directive = route_directive
+        super().__init__(f"Error evaluating condition for route: {route_directive}" if message is None else message)
+
+
+class Router:
+    """
+    The Router class orchestrates the flow of data by evaluating specified route conditions
+    to determine the appropriate route among a set of provided route alternatives.
+
+    To use a Router in Haystack 2.x pipelines we first define a list called routes, where each element
+    is a dictionary representing a route.
+
+    Each route dictionary contains three keys: condition, output, and output_type.
+
+    The condition is a string containing a Python expression that will be evaluated to determine if
+    this route should be selected. The output is a string specifying the name of the output slot for
+    this route, and output_type is a string representation of the expected type of the output.
+
+    After specifying routes list we need to also provide routing variables - variables that will be
+    available for use in the condition expressions within the routes as well as the outputs of the router.
+
+    Example:
+
+        In this example, we create a `Router` instance with two routes.
+        The first route will be selected if the number of streams is less than 2,
+        and will output the `query` variable. The second route will be selected
+        if the number of streams is 2 or more, and will output the `streams` variable.
+        We also specify the routing variables, which are `query` and `streams`. These
+        variables need to be provided in the pipeline `run()` method.
+
+        ```python
+
+            routes = [
+                {"condition": "len(streams) < 2", "output": "query", "output_type": "str"},
+                {"condition": "len(streams) >= 2", "output": "streams", "output_type": "List[ByteStream]"}
+            ]
+
+            router = Router(routes=routes, routing_variables=["query", "streams"])
+        ```
+    """
+
+    def __init__(self, routes: List[Dict], routing_variables: List[str]):
+        """
+        Initialize the Router with a list of routes and the routing variables.
+
+        :param routes: A list of dictionaries, each representing a route with a
+                       boolean condition expression (`condition`), an output slot (`output`),
+                       and the output type as a string representation (`output_type`).
+
+        :param routing_variables: A list of additional pipeline variables that are
+                       used in the boolean condition expressions or as outputs of the router.
+                       These variables should be provided by either the pipeline `run()`
+                       method or by a previous component to the router in the pipeline.
+
+        """
+        self._validate_routes(routes)
+        self._validate_routing_variables(routing_variables)
+        self.routes = routes
+        self.routing_variables = routing_variables
+
+        # Compile the condition expressions for each route
+        self.compiled_expressions = []
+        for route_directive in self.routes:
+            expr_ast = ast.parse(route_directive["condition"], mode="eval")
+            try:
+                compiled_expr = compile(expr_ast, filename="<string>", mode="eval")
+            except Exception as e:
+                raise RouteConditionException(route_directive, "Invalid route condition expression.") from e
+            self.compiled_expressions.append(compiled_expr)
+
+        component.set_input_types(self, **{var: Any for var in routing_variables})
+
+        all_output_types = {}
+        for route in routes:
+            all_output_types.update({route["output"]: route["output_type"]})
+        component.set_output_types(self, **all_output_types)
+
+    def run(self, **kwargs):
+        """
+        Executes the routing logic by evaluating the specified boolean condition expressions
+        for each route in the order they are listed. The method directs the flow
+        of data to the output slot specified in the first route whose expression
+        evaluates to True. If no route's expression evaluates to True, an exception
+        is raised.
+
+        :param kwargs: A dictionary containing the pipeline variables, which should
+                   include all variables listed in `routing_variables`.
+
+        :return: A dictionary containing the output slot and the corresponding result,
+             based on the first route whose expression evaluates to True.
+
+        :raises NoRouteSelectedException: If no route's expression evaluates to True.
+        """
+        local_vars = {context_var: kwargs[context_var] for context_var in self.routing_variables}
+        for i, route_directive in enumerate(self.routes):
+            try:
+                result = eval(self.compiled_expressions[i], local_vars)
path: haystack/preview/components/routers/router.py
text: @user1:
Don't want to be the party pooper, but unless we find an alternative to `eval` I'm not going to approve the PR. With pipelines going around in text format, such a component will always be a security nightmare.
I would rather explore a "declarative" approach, a DSL or whatever we can use to dictate what's the boolean condition... It doesn't need to be as expressive as Python, just usable for the most common use cases. For everything else there's the "custom component" option.

@user2:
+1 on this. We need to ditch this love for `eval`, it's way too dangerous.


@author:
+1 I agree, we'll solve it