Repository: deepset-ai/haystack
Pull request: feat: add custom jinja filter handling to ConditionalRouter
### Related Issues

- fixes #7940 

### Proposed Changes:

Added a `custom_filters: Optional[Dict[str, Callable]]` field to the `ConditionalRouter` constructor to allow users to pass in and use custom Jinja2 filters when creating route conditions.  Also added the same field to `from_dict` to allow users to provide callables when de-serializing a `ConditionalRouter` that was serialized with custom filters.

### How did you test it?

Added 3 unit tests to `test/components/routers/test_conditional_router.py`:
- `test_custom_filter` - ensures a custom test filter works with the router
- `test_custom_filter_serialization` - ensures the router with the custom filter attribute can be serialized and de-serialized
- `test_custom_filter_bad_deserialization` - ensures that a custom exception is generated when a router with custom filers is deserialized without providing callables

### Notes for the reviewer

I updated the docstring with usage information, but not sure what your documentation requirements are otherwise.

### 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:`.
- 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:
Add custom filter support to ConditionalRouter component
*Is your feature request related to a problem? Please describe.**
Currently, the `ConditionalRouter` component doesn't support custom filters in the 'condition' and 'output' fields of a route. This limitation restricts users from applying complex logic or data transformations within these fields, forcing them to pre-process data before it reaches the router or to create additional components for simple transformations. This can lead to less efficient pipelines and reduced flexibility in routing logic.

**Describe the solution you'd like**
We propose adding support for custom filters in the `ConditionalRouter`, similar to how they are implemented in the `OutputAdapter` component. Specifically:

1. Add a 'custom_filters' parameter to the ConditionalRouter's __init__ method, accepting a dictionary of custom filter functions.
2. Allow these custom filters to be used in both the 'condition' and 'output' fields of each route.
3. Implement the necessary logic to apply these filters during the evaluation of conditions and generation of outputs.

This would allow users to write more expressive and powerful routing logic directly within the ConditionalRouter, enhancing its capabilities and flexibility.

**Describe alternatives you've considered**
1. Pre-processing data: Users could transform data before it reaches the `ConditionalRouter`, but this can be cumbersome and less efficient for complex pipelines.
2. Additional components: Creating separate components for data transformation before routing, which can make pipelines more complex and harder to manage.

**Additional context**
This feature would align the `ConditionalRouter` more closely with the `OutputAdapter` in terms of functionality, providing a consistent experience across Haystack components. It would be particularly useful for scenarios involving complex data transformations or when integrating external libraries for specific computations within the routing logic.

Hints from issue discussion:
Hi @vblagoje and team Haystack.  I just implemented a custom component inheriting from `ConditionalRouter` that does this in a recent project.  I'll submit a PR.
PR is up:  https://github.com/deepset-ai/haystack/pull/7957

Commit message: Merge branch 'main' into issue-7940

Patch to review:
diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py
index 70325f4d51..3d675adaed 100644
--- a/haystack/components/routers/conditional_router.py
+++ b/haystack/components/routers/conditional_router.py
@@ -2,7 +2,7 @@
 #
 # SPDX-License-Identifier: Apache-2.0
 
-from typing import Any, Dict, List, Set
+from typing import Any, Callable, Dict, List, Optional, Set
 
 from jinja2 import Environment, TemplateSyntaxError, meta
 from jinja2.nativetypes import NativeEnvironment
@@ -13,6 +13,10 @@
 logger = logging.getLogger(__name__)
 
 
+class BadConditionFilterException(Exception):
+    """Exception raised when a custom filter is not callable."""
+
+
 class NoRouteSelectedException(Exception):
     """Exception raised when no route is selected in ConditionalRouter."""
 
@@ -102,7 +106,7 @@ class ConditionalRouter:
     ```
     """
 
-    def __init__(self, routes: List[Dict]):
+    def __init__(self, routes: List[Dict], custom_filters: Optional[Dict[str, Callable]] = None):
         """
         Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.
 
@@ -113,12 +117,19 @@ def __init__(self, routes: List[Dict]):
             - `output_type`: The type of the output data (e.g., str, List[int]).
             - `output_name`: The name under which the `output` value of the route is published. This name is used to
                 connect the router to other components in the pipeline.
+        :param custom_filters: A dictionary of custom Jinja2 filters to be used in the condition expressions.
+            For example, passing `{"my_filter": my_filter_fcn}` where:
+            - `my_filter` is the name of the custom filter.
+            - `my_filter_fcn` is a callable that takes a variable and returns a value.
+            `{{ my_var|my_filter }}` can then be used inside a route condition expression.
         """
         self._validate_routes(routes)
         self.routes: List[dict] = routes
+        self.custom_filters = custom_filters or {}
 
         # Create a Jinja native environment to inspect variables in the condition templates
         env = NativeEnvironment()
+        env.filters.update(self.custom_filters)
 
         # Inspect the routes to determine input and output types.
         input_types: Set[str] = set()  # let's just store the name, type will always be Any
@@ -146,15 +157,22 @@ def to_dict(self) -> Dict[str, Any]:
             # output_type needs to be serialized to a string
             route["output_type"] = serialize_type(route["output_type"])
 
-        return default_to_dict(self, routes=self.routes)
+        # set all filter callables to None for serialization
+        custom_filters = {filter_name: None for filter_name in self.custom_filters.keys()}
+
+        return default_to_dict(self, routes=self.routes, custom_filters=custom_filters)
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
+    def from_dict(
+        cls, data: Dict[str, Any], custom_filters: Optional[Dict[str, Callable]] = None
+    ) -> "ConditionalRouter":
         """
         Deserializes the component from a dictionary.
 
         :param data:
             The dictionary to deserialize from.
+        :param custom_filters:
+            An optional dictionary of custom Jinja2 filters to be used in the condition expressions.
         :returns:
             The deserialized component.
         """
@@ -163,6 +181,13 @@ def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
         for route in routes:
             # output_type needs to be deserialized from a string to a type
             route["output_type"] = deserialize_type(route["output_type"])
+
+        # Validate and add custom filters to the environment
+        init_params.get("custom_filters", {}).update(custom_filters or {})
+        for filter_name, filter_callable in init_params["custom_filters"].items():
+            if not callable(filter_callable):
+                raise BadConditionFilterException(f"You must specify a callable for custom filter '{filter_name}'.")
+
         return default_from_dict(cls, data)
 
     def run(self, **kwargs):
@@ -185,6 +210,7 @@ def run(self, **kwargs):
         """
         # Create a Jinja native environment to evaluate the condition templates as Python expressions
         env = NativeEnvironment()
+        env.filters.update(self.custom_filters)
 
         for route in self.routes:
             try:
diff --git a/releasenotes/notes/add-custom-filters-to-conditional-router-631eba8bab3c2ae7.yaml b/releasenotes/notes/add-custom-filters-to-conditional-router-631eba8bab3c2ae7.yaml
new file mode 100644
index 0000000000..9f4cdde9f7
--- /dev/null
+++ b/releasenotes/notes/add-custom-filters-to-conditional-router-631eba8bab3c2ae7.yaml
@@ -0,0 +1,6 @@
+---
+features:
+  - |
+    Added custom filters support to ConditionalRouter. Users can now pass in
+    one or more custom Jinja2 filter callables and be able to access those
+    filters when defining condition expressions in routes.
diff --git a/test/components/routers/test_conditional_router.py b/test/components/routers/test_conditional_router.py
index 93d4547841..b887306484 100644
--- a/test/components/routers/test_conditional_router.py
+++ b/test/components/routers/test_conditional_router.py
@@ -8,7 +8,7 @@
 import pytest
 
 from haystack.components.routers import ConditionalRouter
-from haystack.components.routers.conditional_router import NoRouteSelectedException
+from haystack.components.routers.conditional_router import BadConditionFilterException, NoRouteSelectedException
 from haystack.dataclasses import ChatMessage
 
 
@@ -288,3 +288,58 @@ def test_router_serialization_idempotence(self):
         router_dict_first_invocation = copy.deepcopy(router.to_dict())
         router_dict_second_invocation = router.to_dict()
         assert router_dict_first_invocation == router_dict_second_invocation
+
+    def test_custom_filter(self):
+        routes = [
+            {
+                "condition": "{{phone_num|get_area_code == 123}}",
+                "output": "Phone number has a 123 area code",
+                "output_name": "good_phone_num",
+                "output_type": str,
+            },
+            {
+                "condition": "{{phone_num|get_area_code != 123}}",
+                "output": "Phone number does not have 123 area code",
+                "output_name": "bad_phone_num",
+                "output_type": str,
+            },
+        ]
+
+        def get_area_code(phone_num):
+            return int(phone_num.split("-")[0])
+
+        router = ConditionalRouter(routes, custom_filters={"get_area_code": get_area_code})
+        kwargs = {"phone_num": "123-456-7890"}
+        result = router.run(**kwargs)
+        assert result == {"good_phone_num": "Phone number has a 123 area code"}
+        kwargs = {"phone_num": "321-456-7890"}
+        result = router.run(**kwargs)
+        assert result == {"bad_phone_num": "Phone number does not have 123 area code"}
+
+    def test_custom_filter_serialization(self):
+        routes = [{"condition": "{{ test|return_true }}", "output": "true", "output_name": "true", "output_type": str}]
+
+        def return_true(*args):
+            return True
+
+        custom_filters = {"return_true": return_true}
+        router = ConditionalRouter(routes, custom_filters=custom_filters)
+        kwargs = {"test": "test"}
+        result = router.run(**kwargs)
+        assert result == {"true": "true"}
+        serialized_router = router.to_dict()
+        assert serialized_router["init_parameters"]["custom_filters"] == {"return_true": None}
+        deserialized_router = ConditionalRouter.from_dict(serialized_router, custom_filters=custom_filters)
+        assert result == deserialized_router.run(**kwargs)
+
+    def test_custom_filter_bad_deserialization(self):
+        routes = [{"condition": "{{ test|return_true }}", "output": "true", "output_name": "true", "output_type": str}]
+
+        def return_true(*args):
+            return True
+
+        custom_filters = {"return_true": return_true}
+        router = ConditionalRouter(routes, custom_filters=custom_filters)
+        serialized_router = router.to_dict()
+        with pytest.raises(BadConditionFilterException):
+            ConditionalRouter.from_dict(serialized_router)


Code review comments:
diff_hunk: @@ -146,15 +157,22 @@ def to_dict(self) -> Dict[str, Any]:
             # output_type needs to be serialized to a string
             route["output_type"] = serialize_type(route["output_type"])
 
-        return default_to_dict(self, routes=self.routes)
+        # set all filter callables to None for serialization
+        custom_filters = {filter_name: None for filter_name in self.custom_filters.keys()}
path: haystack/components/routers/conditional_router.py
text: @user1:
We should correctly serialize and deserialize callbacks with [these functions](https://github.com/shadeMe/haystack/blob/3e0c532e0e8295d2f258d143b75d12894a9eb38c/haystack/utils/callable_serialization.py#L12).

@author:
Revised all sede operations to match the pattern in OutputAdapter.

diff_hunk: @@ -146,15 +157,22 @@ def to_dict(self) -> Dict[str, Any]:
             # output_type needs to be serialized to a string
             route["output_type"] = serialize_type(route["output_type"])
 
-        return default_to_dict(self, routes=self.routes)
+        # set all filter callables to None for serialization
+        custom_filters = {filter_name: None for filter_name in self.custom_filters.keys()}
+
+        return default_to_dict(self, routes=self.routes, custom_filters=custom_filters)
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
+    def from_dict(
+        cls, data: Dict[str, Any], custom_filters: Optional[Dict[str, Callable]] = None
+    ) -> "ConditionalRouter":
path: haystack/components/routers/conditional_router.py
text: @user1:
We can't modify the signature of the `from_dict` method. 

@author:
Reverted.

diff_hunk: @@ -163,6 +181,13 @@ def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
         for route in routes:
             # output_type needs to be deserialized from a string to a type
             route["output_type"] = deserialize_type(route["output_type"])
+
+        # Validate and add custom filters to the environment
+        init_params.get("custom_filters", {}).update(custom_filters or {})
+        for filter_name, filter_callable in init_params["custom_filters"].items():
+            if not callable(filter_callable):
+                raise BadConditionFilterException(f"You must specify a callable for custom filter '{filter_name}'.")
path: haystack/components/routers/conditional_router.py
text: @user1:
See above for the correct serde procedure.

@author:
Revised per the above.