Repository: deepset-ai/haystack

Issue:
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.



Discussion:
CLAassistant: [![CLA assistant check](https://cla-assistant.io/pull/badge/not_signed)](https://cla-assistant.io/deepset-ai/haystack?pullRequest=7957) <br/>Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our [Contributor License Agreement](https://cla-assistant.io/deepset-ai/haystack?pullRequest=7957) before we can accept your contribution.<br/><sub>You have signed the CLA already but the status is still pending? Let us [recheck](https://cla-assistant.io/check/deepset-ai/haystack?pullRequest=7957) it.</sub>
coveralls: ## Pull Request Test Coverage Report for [Build 9718882629](https://coveralls.io/builds/68375209)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.01%**) to **89.989%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68375209/source?filename=components%2Frouters%2Fconditional_router.py#L224) | 2 | 97.5% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68375209/badge)](https://coveralls.io/builds/68375209) |
| :-- | --: |
| Change from base [Build 9710252897](https://coveralls.io/builds/68358676): |  0.01% |
| Covered Lines: | 6751 |
| Relevant Lines: | 7502 |

---
##### 💛  - [Coveralls](https://coveralls.io)

coveralls: ## Pull Request Test Coverage Report for [Build 9724715471](https://coveralls.io/builds/68376127)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.01%**) to **89.989%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68376127/source?filename=components%2Frouters%2Fconditional_router.py#L224) | 2 | 97.5% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68376127/badge)](https://coveralls.io/builds/68376127) |
| :-- | --: |
| Change from base [Build 9710252897](https://coveralls.io/builds/68358676): |  0.01% |
| Covered Lines: | 6751 |
| Relevant Lines: | 7502 |

---
##### 💛  - [Coveralls](https://coveralls.io)

ChrisPappalardo: Not sure what to do about these integration test errors (all seem related to OpenAI):
![image](https://github.com/deepset-ai/haystack/assets/3485539/bec70d28-e86a-4040-9876-5c689dfaf426)
![image](https://github.com/deepset-ai/haystack/assets/3485539/f0292f00-8dda-4374-ac07-a18a2144519f)

vblagoje: Indeed unrelated, thanks @ChrisPappalardo we'll take a look at integrating this one soon 🙏 
vblagoje: @shadeMe it would be nice to include this one in the upcoming release, if possible 😄 
coveralls: ## Pull Request Test Coverage Report for [Build 9740569615](https://coveralls.io/builds/68388820)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.01%**) to **89.989%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68388820/source?filename=components%2Frouters%2Fconditional_router.py#L224) | 2 | 97.5% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68388820/badge)](https://coveralls.io/builds/68388820) |
| :-- | --: |
| Change from base [Build 9740565126](https://coveralls.io/builds/68388803): |  0.01% |
| Covered Lines: | 6751 |
| Relevant Lines: | 7502 |

---
##### 💛  - [Coveralls](https://coveralls.io)

coveralls: ## Pull Request Test Coverage Report for [Build 9768491597](https://coveralls.io/builds/68438806)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.008%**) to **89.976%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68438806/source?filename=components%2Frouters%2Fconditional_router.py#L209) | 2 | 97.4% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68438806/badge)](https://coveralls.io/builds/68438806) |
| :-- | --: |
| Change from base [Build 9758508178](https://coveralls.io/builds/68415918): |  0.008% |
| Covered Lines: | 6750 |
| Relevant Lines: | 7502 |

---
##### 💛  - [Coveralls](https://coveralls.io)

coveralls: ## Pull Request Test Coverage Report for [Build 9779774624](https://coveralls.io/builds/68447500)

### Warning: This coverage report may be inaccurate.

This pull request's base commit is no longer the HEAD commit of its target branch. This means it includes changes from outside the original pull request, including, potentially, unrelated coverage changes.

- For more information on this, see <a target="_blank" href="https://docs.coveralls.io/build-types#tracking-coverage-changes-with-pull_request-builds">Tracking coverage changes with pull request builds</a>.
- To avoid this issue with future PRs, see these <a target="_blank" href="https://docs.coveralls.io/build-types#recommended-ci-configurations">Recommended CI Configurations</a>.
- For a quick fix, <a target="_blank" href="https://github.blog/changelog/2022-02-03-more-ways-to-keep-your-pull-request-branch-up-to-date/#update-your-pull-request-branch-by-rebasing">rebase this PR at GitHub</a>. Your next report should be accurate.

### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.03%**) to **89.997%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68447500/source?filename=components%2Frouters%2Fconditional_router.py#L209) | 2 | 97.4% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68447500/badge)](https://coveralls.io/builds/68447500) |
| :-- | --: |
| Change from base [Build 9758508178](https://coveralls.io/builds/68415918): |  0.03% |
| Covered Lines: | 6766 |
| Relevant Lines: | 7518 |

---
##### 💛  - [Coveralls](https://coveralls.io)

coveralls: ## Pull Request Test Coverage Report for [Build 9782596228](https://coveralls.io/builds/68454945)


### Details

* **0** of **0**   changed or added relevant lines in **0** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.008%**) to **89.997%**

---


|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [components/routers/conditional_router.py](https://coveralls.io/builds/68454945/source?filename=components%2Frouters%2Fconditional_router.py#L209) | 2 | 97.4% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/68454945/badge)](https://coveralls.io/builds/68454945) |
| :-- | --: |
| Change from base [Build 9777342637](https://coveralls.io/builds/68443645): |  0.008% |
| Covered Lines: | 6766 |
| Relevant Lines: | 7518 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/haystack/components/joiners/document_joiner.py b/haystack/components/joiners/document_joiner.py
index 6e1e5a80..bd51613d 100644
--- a/haystack/components/joiners/document_joiner.py
+++ b/haystack/components/joiners/document_joiner.py
@@ -195,9 +195,11 @@ class DocumentJoiner:
             std_dev = (sum((x - mean_score) ** 2 for x in scores_list) / len(scores_list)) ** 0.5
             min_score = mean_score - 3 * std_dev
             max_score = mean_score + 3 * std_dev
+            delta_score = max_score - min_score
 
             for doc in documents:
-                doc.score = (doc.score - min_score) / (max_score - min_score)
+                doc.score = (doc.score - min_score) / delta_score if delta_score != 0.0 else 0.0
+                # if all docs have the same score delta_score is 0, the docs are uninformative for the query
 
         output = self._concatenate(document_lists=document_lists)
 
diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py
index 70325f4d..be72c7cc 100644
--- a/haystack/components/routers/conditional_router.py
+++ b/haystack/components/routers/conditional_router.py
@@ -2,13 +2,13 @@
 #
 # 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
 
 from haystack import component, default_from_dict, default_to_dict, logging
-from haystack.utils import deserialize_type, serialize_type
+from haystack.utils import deserialize_callable, deserialize_type, serialize_callable, serialize_type
 
 logger = logging.getLogger(__name__)
 
@@ -102,7 +102,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 +113,20 @@ class ConditionalRouter:
             - `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 `my_var:str` and returns `my_var[:3]`.
+              `{{ my_var|my_filter }}` can then be used inside a route condition expression like so:
+                `"condition": "{{ my_var|my_filter == 'foo' }}"`.
         """
         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
@@ -145,8 +153,8 @@ class ConditionalRouter:
         for route in self.routes:
             # 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)
+        se_filters = {name: serialize_callable(filter_func) for name, filter_func in self.custom_filters.items()}
+        return default_to_dict(self, routes=self.routes, custom_filters=se_filters)
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
@@ -163,6 +171,8 @@ class 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"])
+        for name, filter_func in init_params.get("custom_filters", {}).items():
+            init_params["custom_filters"][name] = deserialize_callable(filter_func) if filter_func else None
         return default_from_dict(cls, data)
 
     def run(self, **kwargs):
@@ -185,6 +195,7 @@ class ConditionalRouter:
         """
         # 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 00000000..9f4cdde9
--- /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.


Test patch:
diff --git a/test/components/joiners/test_document_joiner.py b/test/components/joiners/test_document_joiner.py
index 41fe2755..0da08247 100644
--- a/test/components/joiners/test_document_joiner.py
+++ b/test/components/joiners/test_document_joiner.py
@@ -145,6 +145,36 @@ class TestDocumentJoiner:
         ]
         assert all(doc.id in expected_document_ids for doc in output["documents"])
 
+    def test_run_with_distribution_based_rank_fusion_join_mode_same_scores(self):
+        joiner = DocumentJoiner(join_mode="distribution_based_rank_fusion")
+        documents_1 = [
+            Document(content="a", score=0.2),
+            Document(content="b", score=0.2),
+            Document(content="c", score=0.2),
+        ]
+        documents_2 = [
+            Document(content="d", score=0.5),
+            Document(content="e", score=0.8),
+            Document(content="f", score=1.1, meta={"key": "value"}),
+            Document(content="g", score=0.3),
+            Document(content="a", score=0.3),
+        ]
+        output = joiner.run([documents_1, documents_2])
+        assert len(output["documents"]) == 7
+        expected_document_ids = [
+            doc.id
+            for doc in [
+                Document(content="a", score=0),
+                Document(content="b", score=0),
+                Document(content="c", score=0),
+                Document(content="d", score=0.44),
+                Document(content="e", score=0.60),
+                Document(content="f", score=0.76, meta={"key": "value"}),
+                Document(content="g", score=0.33),
+            ]
+        ]
+        assert all(doc.id in expected_document_ids for doc in output["documents"])
+
     def test_run_with_top_k_in_run_method(self):
         joiner = DocumentJoiner()
         documents_1 = [Document(content="a"), Document(content="b"), Document(content="c")]
diff --git a/test/components/routers/test_conditional_router.py b/test/components/routers/test_conditional_router.py
index 93d45478..0b905957 100644
--- a/test/components/routers/test_conditional_router.py
+++ b/test/components/routers/test_conditional_router.py
@@ -12,6 +12,11 @@ from haystack.components.routers.conditional_router import NoRouteSelectedExcept
 from haystack.dataclasses import ChatMessage
 
 
+def custom_filter_to_sede(value):
+    """splits by hyphen and returns the first part"""
+    return int(value.split("-")[0])
+
+
 class TestRouter:
     @pytest.fixture
     def routes(self):
@@ -288,3 +293,47 @@ class TestRouter:
         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,
+            },
+        ]
+
+        router = ConditionalRouter(routes, custom_filters={"get_area_code": custom_filter_to_sede})
+        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_sede_with_custom_filter(self):
+        routes = [
+            {
+                "condition": "{{ test|custom_filter_to_sede == 123 }}",
+                "output": "123",
+                "output_name": "test",
+                "output_type": int,
+            }
+        ]
+        custom_filters = {"custom_filter_to_sede": custom_filter_to_sede}
+        router = ConditionalRouter(routes, custom_filters=custom_filters)
+        kwargs = {"test": "123-456-789"}
+        result = router.run(**kwargs)
+        assert result == {"test": 123}
+        serialized_router = router.to_dict()
+        deserialized_router = ConditionalRouter.from_dict(serialized_router)
+        assert deserialized_router.custom_filters == router.custom_filters
+        assert deserialized_router.custom_filters["custom_filter_to_sede"]("123-456-789") == 123
+        assert result == deserialized_router.run(**kwargs)