We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN 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.



--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `haystack/components/routers/conditional_router.py`
Content:
```
1 # SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
2 #
3 # SPDX-License-Identifier: Apache-2.0
4 
5 from typing import Any, Dict, List, Set
6 
7 from jinja2 import Environment, TemplateSyntaxError, meta
8 from jinja2.nativetypes import NativeEnvironment
9 
10 from haystack import component, default_from_dict, default_to_dict, logging
11 from haystack.utils import deserialize_type, serialize_type
12 
13 logger = logging.getLogger(__name__)
14 
15 
16 class NoRouteSelectedException(Exception):
17     """Exception raised when no route is selected in ConditionalRouter."""
18 
19 
20 class RouteConditionException(Exception):
21     """Exception raised when there is an error parsing or evaluating the condition expression in ConditionalRouter."""
22 
23 
24 @component
25 class ConditionalRouter:
26     """
27     `ConditionalRouter` allows data routing based on specific conditions.
28 
29     This is achieved by defining a list named `routes`. Each element in this list is a dictionary representing a
30     single route.
31     A route dictionary comprises four key elements:
32     - `condition`: A Jinja2 string expression that determines if the route is selected.
33     - `output`: A Jinja2 expression defining the route's output value.
34     - `output_type`: The type of the output data (e.g., `str`, `List[int]`).
35     - `output_name`: The name under which the `output` value of the route is published. This name is used to connect
36     the router to other components in the pipeline.
37 
38     Usage example:
39     ```python
40     from typing import List
41     from haystack.components.routers import ConditionalRouter
42 
43     routes = [
44         {
45             "condition": "{{streams|length > 2}}",
46             "output": "{{streams}}",
47             "output_name": "enough_streams",
48             "output_type": List[int],
49         },
50         {
51             "condition": "{{streams|length <= 2}}",
52             "output": "{{streams}}",
53             "output_name": "insufficient_streams",
54             "output_type": List[int],
55         },
56     ]
57     router = ConditionalRouter(routes)
58     # When 'streams' has more than 2 items, 'enough_streams' output will activate, emitting the list [1, 2, 3]
59     kwargs = {"streams": [1, 2, 3], "query": "Haystack"}
60     result = router.run(**kwargs)
61     assert result == {"enough_streams": [1, 2, 3]}
62     ```
63 
64     In this example, we configure two routes. The first route sends the 'streams' value to 'enough_streams' if the
65     stream count exceeds two. Conversely, the second route directs 'streams' to 'insufficient_streams' when there
66     are two or fewer streams.
67 
68     In the pipeline setup, the router is connected to other components using the output names. For example, the
69     'enough_streams' output might be connected to another component that processes the streams, while the
70     'insufficient_streams' output might be connected to a component that fetches more streams, and so on.
71 
72 
73     Here is a pseudocode example of a pipeline that uses the `ConditionalRouter` and routes fetched `ByteStreams` to
74     different components depending on the number of streams fetched:
75     ```python
76     from typing import List
77     from haystack import Pipeline
78     from haystack.dataclasses import ByteStream
79     from haystack.components.routers import ConditionalRouter
80 
81     routes = [
82         {
83             "condition": "{{streams|length > 2}}",
84             "output": "{{streams}}",
85             "output_name": "enough_streams",
86             "output_type": List[ByteStream],
87         },
88         {
89             "condition": "{{streams|length <= 2}}",
90             "output": "{{streams}}",
91             "output_name": "insufficient_streams",
92             "output_type": List[ByteStream],
93         },
94     ]
95 
96     pipe = Pipeline()
97     pipe.add_component("router", router)
98     ...
99     pipe.connect("router.enough_streams", "some_component_a.streams")
100     pipe.connect("router.insufficient_streams", "some_component_b.streams_or_some_other_input")
101     ...
102     ```
103     """
104 
105     def __init__(self, routes: List[Dict]):
106         """
107         Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.
108 
109         :param routes: A list of dictionaries, each defining a route.
110             A route dictionary comprises four key elements:
111             - `condition`: A Jinja2 string expression that determines if the route is selected.
112             - `output`: A Jinja2 expression defining the route's output value.
113             - `output_type`: The type of the output data (e.g., str, List[int]).
114             - `output_name`: The name under which the `output` value of the route is published. This name is used to
115                 connect the router to other components in the pipeline.
116         """
117         self._validate_routes(routes)
118         self.routes: List[dict] = routes
119 
120         # Create a Jinja native environment to inspect variables in the condition templates
121         env = NativeEnvironment()
122 
123         # Inspect the routes to determine input and output types.
124         input_types: Set[str] = set()  # let's just store the name, type will always be Any
125         output_types: Dict[str, str] = {}
126 
127         for route in routes:
128             # extract inputs
129             route_input_names = self._extract_variables(env, [route["output"], route["condition"]])
130             input_types.update(route_input_names)
131 
132             # extract outputs
133             output_types.update({route["output_name"]: route["output_type"]})
134 
135         component.set_input_types(self, **{var: Any for var in input_types})
136         component.set_output_types(self, **output_types)
137 
138     def to_dict(self) -> Dict[str, Any]:
139         """
140         Serializes the component to a dictionary.
141 
142         :returns:
143             Dictionary with serialized data.
144         """
145         for route in self.routes:
146             # output_type needs to be serialized to a string
147             route["output_type"] = serialize_type(route["output_type"])
148 
149         return default_to_dict(self, routes=self.routes)
150 
151     @classmethod
152     def from_dict(cls, data: Dict[str, Any]) -> "ConditionalRouter":
153         """
154         Deserializes the component from a dictionary.
155 
156         :param data:
157             The dictionary to deserialize from.
158         :returns:
159             The deserialized component.
160         """
161         init_params = data.get("init_parameters", {})
162         routes = init_params.get("routes")
163         for route in routes:
164             # output_type needs to be deserialized from a string to a type
165             route["output_type"] = deserialize_type(route["output_type"])
166         return default_from_dict(cls, data)
167 
168     def run(self, **kwargs):
169         """
170         Executes the routing logic.
171 
172         Executes the routing logic by evaluating the specified boolean condition expressions for each route in the
173         order they are listed. The method directs the flow of data to the output specified in the first route whose
174         `condition` is True.
175 
176         :param kwargs: All variables used in the `condition` expressed in the routes. When the component is used in a
177             pipeline, these variables are passed from the previous component's output.
178 
179         :returns: A dictionary where the key is the `output_name` of the selected route and the value is the `output`
180             of the selected route.
181 
182         :raises NoRouteSelectedException: If no `condition' in the routes is `True`.
183         :raises RouteConditionException: If there is an error parsing or evaluating the `condition` expression in the
184             routes.
185         """
186         # Create a Jinja native environment to evaluate the condition templates as Python expressions
187         env = NativeEnvironment()
188 
189         for route in self.routes:
190             try:
191                 t = env.from_string(route["condition"])
192                 if t.render(**kwargs):
193                     # We now evaluate the `output` expression to determine the route output
194                     t_output = env.from_string(route["output"])
195                     output = t_output.render(**kwargs)
196                     # and return the output as a dictionary under the output_name key
197                     return {route["output_name"]: output}
198             except Exception as e:
199                 raise RouteConditionException(f"Error evaluating condition for route '{route}': {e}") from e
200 
201         raise NoRouteSelectedException(f"No route fired. Routes: {self.routes}")
202 
203     def _validate_routes(self, routes: List[Dict]):
204         """
205         Validates a list of routes.
206 
207         :param routes: A list of routes.
208         """
209         env = NativeEnvironment()
210         for route in routes:
211             try:
212                 keys = set(route.keys())
213             except AttributeError:
214                 raise ValueError(f"Route must be a dictionary, got: {route}")
215 
216             mandatory_fields = {"condition", "output", "output_type", "output_name"}
217             has_all_mandatory_fields = mandatory_fields.issubset(keys)
218             if not has_all_mandatory_fields:
219                 raise ValueError(
220                     f"Route must contain 'condition', 'output', 'output_type' and 'output_name' fields: {route}"
221                 )
222             for field in ["condition", "output"]:
223                 if not self._validate_template(env, route[field]):
224                     raise ValueError(f"Invalid template for field '{field}': {route[field]}")
225 
226     def _extract_variables(self, env: NativeEnvironment, templates: List[str]) -> Set[str]:
227         """
228         Extracts all variables from a list of Jinja template strings.
229 
230         :param env: A Jinja environment.
231         :param templates: A list of Jinja template strings.
232         :returns: A set of variable names.
233         """
234         variables = set()
235         for template in templates:
236             ast = env.parse(template)
237             variables.update(meta.find_undeclared_variables(ast))
238         return variables
239 
240     def _validate_template(self, env: Environment, template_text: str):
241         """
242         Validates a template string by parsing it with Jinja.
243 
244         :param env: A Jinja environment.
245         :param template_text: A Jinja template string.
246         :returns: `True` if the template is valid, `False` otherwise.
247         """
248         try:
249             env.parse(template_text)
250             return True
251         except TemplateSyntaxError:
252             return False
253
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py
--- 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 @@
     ```
     """
 
-    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 @@
             - `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 @@
         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 @@
         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 @@
         """
         # 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:

{"golden_diff": "diff --git a/haystack/components/routers/conditional_router.py b/haystack/components/routers/conditional_router.py\n--- a/haystack/components/routers/conditional_router.py\n+++ b/haystack/components/routers/conditional_router.py\n@@ -2,13 +2,13 @@\n #\n # SPDX-License-Identifier: Apache-2.0\n \n-from typing import Any, Dict, List, Set\n+from typing import Any, Callable, Dict, List, Optional, Set\n \n from jinja2 import Environment, TemplateSyntaxError, meta\n from jinja2.nativetypes import NativeEnvironment\n \n from haystack import component, default_from_dict, default_to_dict, logging\n-from haystack.utils import deserialize_type, serialize_type\n+from haystack.utils import deserialize_callable, deserialize_type, serialize_callable, serialize_type\n \n logger = logging.getLogger(__name__)\n \n@@ -102,7 +102,7 @@\n     ```\n     \"\"\"\n \n-    def __init__(self, routes: List[Dict]):\n+    def __init__(self, routes: List[Dict], custom_filters: Optional[Dict[str, Callable]] = None):\n         \"\"\"\n         Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.\n \n@@ -113,12 +113,20 @@\n             - `output_type`: The type of the output data (e.g., str, List[int]).\n             - `output_name`: The name under which the `output` value of the route is published. This name is used to\n                 connect the router to other components in the pipeline.\n+        :param custom_filters: A dictionary of custom Jinja2 filters to be used in the condition expressions.\n+            For example, passing `{\"my_filter\": my_filter_fcn}` where:\n+            - `my_filter` is the name of the custom filter.\n+            - `my_filter_fcn` is a callable that takes `my_var:str` and returns `my_var[:3]`.\n+              `{{ my_var|my_filter }}` can then be used inside a route condition expression like so:\n+                `\"condition\": \"{{ my_var|my_filter == 'foo' }}\"`.\n         \"\"\"\n         self._validate_routes(routes)\n         self.routes: List[dict] = routes\n+        self.custom_filters = custom_filters or {}\n \n         # Create a Jinja native environment to inspect variables in the condition templates\n         env = NativeEnvironment()\n+        env.filters.update(self.custom_filters)\n \n         # Inspect the routes to determine input and output types.\n         input_types: Set[str] = set()  # let's just store the name, type will always be Any\n@@ -145,8 +153,8 @@\n         for route in self.routes:\n             # output_type needs to be serialized to a string\n             route[\"output_type\"] = serialize_type(route[\"output_type\"])\n-\n-        return default_to_dict(self, routes=self.routes)\n+        se_filters = {name: serialize_callable(filter_func) for name, filter_func in self.custom_filters.items()}\n+        return default_to_dict(self, routes=self.routes, custom_filters=se_filters)\n \n     @classmethod\n     def from_dict(cls, data: Dict[str, Any]) -> \"ConditionalRouter\":\n@@ -163,6 +171,8 @@\n         for route in routes:\n             # output_type needs to be deserialized from a string to a type\n             route[\"output_type\"] = deserialize_type(route[\"output_type\"])\n+        for name, filter_func in init_params.get(\"custom_filters\", {}).items():\n+            init_params[\"custom_filters\"][name] = deserialize_callable(filter_func) if filter_func else None\n         return default_from_dict(cls, data)\n \n     def run(self, **kwargs):\n@@ -185,6 +195,7 @@\n         \"\"\"\n         # Create a Jinja native environment to evaluate the condition templates as Python expressions\n         env = NativeEnvironment()\n+        env.filters.update(self.custom_filters)\n \n         for route in self.routes:\n             try:\n", "issue": "Add custom filter support to ConditionalRouter component\n*Is your feature request related to a problem? Please describe.**\r\nCurrently, 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.\r\n\r\n**Describe the solution you'd like**\r\nWe propose adding support for custom filters in the `ConditionalRouter`, similar to how they are implemented in the `OutputAdapter` component. Specifically:\r\n\r\n1. Add a 'custom_filters' parameter to the ConditionalRouter's __init__ method, accepting a dictionary of custom filter functions.\r\n2. Allow these custom filters to be used in both the 'condition' and 'output' fields of each route.\r\n3. Implement the necessary logic to apply these filters during the evaluation of conditions and generation of outputs.\r\n\r\nThis would allow users to write more expressive and powerful routing logic directly within the ConditionalRouter, enhancing its capabilities and flexibility.\r\n\r\n**Describe alternatives you've considered**\r\n1. Pre-processing data: Users could transform data before it reaches the `ConditionalRouter`, but this can be cumbersome and less efficient for complex pipelines.\r\n2. Additional components: Creating separate components for data transformation before routing, which can make pipelines more complex and harder to manage.\r\n\r\n**Additional context**\r\nThis 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.\r\n\r\n\n", "before_files": [{"content": "# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>\n#\n# SPDX-License-Identifier: Apache-2.0\n\nfrom typing import Any, Dict, List, Set\n\nfrom jinja2 import Environment, TemplateSyntaxError, meta\nfrom jinja2.nativetypes import NativeEnvironment\n\nfrom haystack import component, default_from_dict, default_to_dict, logging\nfrom haystack.utils import deserialize_type, serialize_type\n\nlogger = logging.getLogger(__name__)\n\n\nclass NoRouteSelectedException(Exception):\n    \"\"\"Exception raised when no route is selected in ConditionalRouter.\"\"\"\n\n\nclass RouteConditionException(Exception):\n    \"\"\"Exception raised when there is an error parsing or evaluating the condition expression in ConditionalRouter.\"\"\"\n\n\n@component\nclass ConditionalRouter:\n    \"\"\"\n    `ConditionalRouter` allows data routing based on specific conditions.\n\n    This is achieved by defining a list named `routes`. Each element in this list is a dictionary representing a\n    single route.\n    A route dictionary comprises four key elements:\n    - `condition`: A Jinja2 string expression that determines if the route is selected.\n    - `output`: A Jinja2 expression defining the route's output value.\n    - `output_type`: The type of the output data (e.g., `str`, `List[int]`).\n    - `output_name`: The name under which the `output` value of the route is published. This name is used to connect\n    the router to other components in the pipeline.\n\n    Usage example:\n    ```python\n    from typing import List\n    from haystack.components.routers import ConditionalRouter\n\n    routes = [\n        {\n            \"condition\": \"{{streams|length > 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"enough_streams\",\n            \"output_type\": List[int],\n        },\n        {\n            \"condition\": \"{{streams|length <= 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"insufficient_streams\",\n            \"output_type\": List[int],\n        },\n    ]\n    router = ConditionalRouter(routes)\n    # When 'streams' has more than 2 items, 'enough_streams' output will activate, emitting the list [1, 2, 3]\n    kwargs = {\"streams\": [1, 2, 3], \"query\": \"Haystack\"}\n    result = router.run(**kwargs)\n    assert result == {\"enough_streams\": [1, 2, 3]}\n    ```\n\n    In this example, we configure two routes. The first route sends the 'streams' value to 'enough_streams' if the\n    stream count exceeds two. Conversely, the second route directs 'streams' to 'insufficient_streams' when there\n    are two or fewer streams.\n\n    In the pipeline setup, the router is connected to other components using the output names. For example, the\n    'enough_streams' output might be connected to another component that processes the streams, while the\n    'insufficient_streams' output might be connected to a component that fetches more streams, and so on.\n\n\n    Here is a pseudocode example of a pipeline that uses the `ConditionalRouter` and routes fetched `ByteStreams` to\n    different components depending on the number of streams fetched:\n    ```python\n    from typing import List\n    from haystack import Pipeline\n    from haystack.dataclasses import ByteStream\n    from haystack.components.routers import ConditionalRouter\n\n    routes = [\n        {\n            \"condition\": \"{{streams|length > 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"enough_streams\",\n            \"output_type\": List[ByteStream],\n        },\n        {\n            \"condition\": \"{{streams|length <= 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"insufficient_streams\",\n            \"output_type\": List[ByteStream],\n        },\n    ]\n\n    pipe = Pipeline()\n    pipe.add_component(\"router\", router)\n    ...\n    pipe.connect(\"router.enough_streams\", \"some_component_a.streams\")\n    pipe.connect(\"router.insufficient_streams\", \"some_component_b.streams_or_some_other_input\")\n    ...\n    ```\n    \"\"\"\n\n    def __init__(self, routes: List[Dict]):\n        \"\"\"\n        Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.\n\n        :param routes: A list of dictionaries, each defining a route.\n            A route dictionary comprises four key elements:\n            - `condition`: A Jinja2 string expression that determines if the route is selected.\n            - `output`: A Jinja2 expression defining the route's output value.\n            - `output_type`: The type of the output data (e.g., str, List[int]).\n            - `output_name`: The name under which the `output` value of the route is published. This name is used to\n                connect the router to other components in the pipeline.\n        \"\"\"\n        self._validate_routes(routes)\n        self.routes: List[dict] = routes\n\n        # Create a Jinja native environment to inspect variables in the condition templates\n        env = NativeEnvironment()\n\n        # Inspect the routes to determine input and output types.\n        input_types: Set[str] = set()  # let's just store the name, type will always be Any\n        output_types: Dict[str, str] = {}\n\n        for route in routes:\n            # extract inputs\n            route_input_names = self._extract_variables(env, [route[\"output\"], route[\"condition\"]])\n            input_types.update(route_input_names)\n\n            # extract outputs\n            output_types.update({route[\"output_name\"]: route[\"output_type\"]})\n\n        component.set_input_types(self, **{var: Any for var in input_types})\n        component.set_output_types(self, **output_types)\n\n    def to_dict(self) -> Dict[str, Any]:\n        \"\"\"\n        Serializes the component to a dictionary.\n\n        :returns:\n            Dictionary with serialized data.\n        \"\"\"\n        for route in self.routes:\n            # output_type needs to be serialized to a string\n            route[\"output_type\"] = serialize_type(route[\"output_type\"])\n\n        return default_to_dict(self, routes=self.routes)\n\n    @classmethod\n    def from_dict(cls, data: Dict[str, Any]) -> \"ConditionalRouter\":\n        \"\"\"\n        Deserializes the component from a dictionary.\n\n        :param data:\n            The dictionary to deserialize from.\n        :returns:\n            The deserialized component.\n        \"\"\"\n        init_params = data.get(\"init_parameters\", {})\n        routes = init_params.get(\"routes\")\n        for route in routes:\n            # output_type needs to be deserialized from a string to a type\n            route[\"output_type\"] = deserialize_type(route[\"output_type\"])\n        return default_from_dict(cls, data)\n\n    def run(self, **kwargs):\n        \"\"\"\n        Executes the routing logic.\n\n        Executes the routing logic by evaluating the specified boolean condition expressions for each route in the\n        order they are listed. The method directs the flow of data to the output specified in the first route whose\n        `condition` is True.\n\n        :param kwargs: All variables used in the `condition` expressed in the routes. When the component is used in a\n            pipeline, these variables are passed from the previous component's output.\n\n        :returns: A dictionary where the key is the `output_name` of the selected route and the value is the `output`\n            of the selected route.\n\n        :raises NoRouteSelectedException: If no `condition' in the routes is `True`.\n        :raises RouteConditionException: If there is an error parsing or evaluating the `condition` expression in the\n            routes.\n        \"\"\"\n        # Create a Jinja native environment to evaluate the condition templates as Python expressions\n        env = NativeEnvironment()\n\n        for route in self.routes:\n            try:\n                t = env.from_string(route[\"condition\"])\n                if t.render(**kwargs):\n                    # We now evaluate the `output` expression to determine the route output\n                    t_output = env.from_string(route[\"output\"])\n                    output = t_output.render(**kwargs)\n                    # and return the output as a dictionary under the output_name key\n                    return {route[\"output_name\"]: output}\n            except Exception as e:\n                raise RouteConditionException(f\"Error evaluating condition for route '{route}': {e}\") from e\n\n        raise NoRouteSelectedException(f\"No route fired. Routes: {self.routes}\")\n\n    def _validate_routes(self, routes: List[Dict]):\n        \"\"\"\n        Validates a list of routes.\n\n        :param routes: A list of routes.\n        \"\"\"\n        env = NativeEnvironment()\n        for route in routes:\n            try:\n                keys = set(route.keys())\n            except AttributeError:\n                raise ValueError(f\"Route must be a dictionary, got: {route}\")\n\n            mandatory_fields = {\"condition\", \"output\", \"output_type\", \"output_name\"}\n            has_all_mandatory_fields = mandatory_fields.issubset(keys)\n            if not has_all_mandatory_fields:\n                raise ValueError(\n                    f\"Route must contain 'condition', 'output', 'output_type' and 'output_name' fields: {route}\"\n                )\n            for field in [\"condition\", \"output\"]:\n                if not self._validate_template(env, route[field]):\n                    raise ValueError(f\"Invalid template for field '{field}': {route[field]}\")\n\n    def _extract_variables(self, env: NativeEnvironment, templates: List[str]) -> Set[str]:\n        \"\"\"\n        Extracts all variables from a list of Jinja template strings.\n\n        :param env: A Jinja environment.\n        :param templates: A list of Jinja template strings.\n        :returns: A set of variable names.\n        \"\"\"\n        variables = set()\n        for template in templates:\n            ast = env.parse(template)\n            variables.update(meta.find_undeclared_variables(ast))\n        return variables\n\n    def _validate_template(self, env: Environment, template_text: str):\n        \"\"\"\n        Validates a template string by parsing it with Jinja.\n\n        :param env: A Jinja environment.\n        :param template_text: A Jinja template string.\n        :returns: `True` if the template is valid, `False` otherwise.\n        \"\"\"\n        try:\n            env.parse(template_text)\n            return True\n        except TemplateSyntaxError:\n            return False\n", "path": "haystack/components/routers/conditional_router.py"}], "after_files": [{"content": "# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>\n#\n# SPDX-License-Identifier: Apache-2.0\n\nfrom typing import Any, Callable, Dict, List, Optional, Set\n\nfrom jinja2 import Environment, TemplateSyntaxError, meta\nfrom jinja2.nativetypes import NativeEnvironment\n\nfrom haystack import component, default_from_dict, default_to_dict, logging\nfrom haystack.utils import deserialize_callable, deserialize_type, serialize_callable, serialize_type\n\nlogger = logging.getLogger(__name__)\n\n\nclass NoRouteSelectedException(Exception):\n    \"\"\"Exception raised when no route is selected in ConditionalRouter.\"\"\"\n\n\nclass RouteConditionException(Exception):\n    \"\"\"Exception raised when there is an error parsing or evaluating the condition expression in ConditionalRouter.\"\"\"\n\n\n@component\nclass ConditionalRouter:\n    \"\"\"\n    `ConditionalRouter` allows data routing based on specific conditions.\n\n    This is achieved by defining a list named `routes`. Each element in this list is a dictionary representing a\n    single route.\n    A route dictionary comprises four key elements:\n    - `condition`: A Jinja2 string expression that determines if the route is selected.\n    - `output`: A Jinja2 expression defining the route's output value.\n    - `output_type`: The type of the output data (e.g., `str`, `List[int]`).\n    - `output_name`: The name under which the `output` value of the route is published. This name is used to connect\n    the router to other components in the pipeline.\n\n    Usage example:\n    ```python\n    from typing import List\n    from haystack.components.routers import ConditionalRouter\n\n    routes = [\n        {\n            \"condition\": \"{{streams|length > 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"enough_streams\",\n            \"output_type\": List[int],\n        },\n        {\n            \"condition\": \"{{streams|length <= 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"insufficient_streams\",\n            \"output_type\": List[int],\n        },\n    ]\n    router = ConditionalRouter(routes)\n    # When 'streams' has more than 2 items, 'enough_streams' output will activate, emitting the list [1, 2, 3]\n    kwargs = {\"streams\": [1, 2, 3], \"query\": \"Haystack\"}\n    result = router.run(**kwargs)\n    assert result == {\"enough_streams\": [1, 2, 3]}\n    ```\n\n    In this example, we configure two routes. The first route sends the 'streams' value to 'enough_streams' if the\n    stream count exceeds two. Conversely, the second route directs 'streams' to 'insufficient_streams' when there\n    are two or fewer streams.\n\n    In the pipeline setup, the router is connected to other components using the output names. For example, the\n    'enough_streams' output might be connected to another component that processes the streams, while the\n    'insufficient_streams' output might be connected to a component that fetches more streams, and so on.\n\n\n    Here is a pseudocode example of a pipeline that uses the `ConditionalRouter` and routes fetched `ByteStreams` to\n    different components depending on the number of streams fetched:\n    ```python\n    from typing import List\n    from haystack import Pipeline\n    from haystack.dataclasses import ByteStream\n    from haystack.components.routers import ConditionalRouter\n\n    routes = [\n        {\n            \"condition\": \"{{streams|length > 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"enough_streams\",\n            \"output_type\": List[ByteStream],\n        },\n        {\n            \"condition\": \"{{streams|length <= 2}}\",\n            \"output\": \"{{streams}}\",\n            \"output_name\": \"insufficient_streams\",\n            \"output_type\": List[ByteStream],\n        },\n    ]\n\n    pipe = Pipeline()\n    pipe.add_component(\"router\", router)\n    ...\n    pipe.connect(\"router.enough_streams\", \"some_component_a.streams\")\n    pipe.connect(\"router.insufficient_streams\", \"some_component_b.streams_or_some_other_input\")\n    ...\n    ```\n    \"\"\"\n\n    def __init__(self, routes: List[Dict], custom_filters: Optional[Dict[str, Callable]] = None):\n        \"\"\"\n        Initializes the `ConditionalRouter` with a list of routes detailing the conditions for routing.\n\n        :param routes: A list of dictionaries, each defining a route.\n            A route dictionary comprises four key elements:\n            - `condition`: A Jinja2 string expression that determines if the route is selected.\n            - `output`: A Jinja2 expression defining the route's output value.\n            - `output_type`: The type of the output data (e.g., str, List[int]).\n            - `output_name`: The name under which the `output` value of the route is published. This name is used to\n                connect the router to other components in the pipeline.\n        :param custom_filters: A dictionary of custom Jinja2 filters to be used in the condition expressions.\n            For example, passing `{\"my_filter\": my_filter_fcn}` where:\n            - `my_filter` is the name of the custom filter.\n            - `my_filter_fcn` is a callable that takes `my_var:str` and returns `my_var[:3]`.\n              `{{ my_var|my_filter }}` can then be used inside a route condition expression like so:\n                `\"condition\": \"{{ my_var|my_filter == 'foo' }}\"`.\n        \"\"\"\n        self._validate_routes(routes)\n        self.routes: List[dict] = routes\n        self.custom_filters = custom_filters or {}\n\n        # Create a Jinja native environment to inspect variables in the condition templates\n        env = NativeEnvironment()\n        env.filters.update(self.custom_filters)\n\n        # Inspect the routes to determine input and output types.\n        input_types: Set[str] = set()  # let's just store the name, type will always be Any\n        output_types: Dict[str, str] = {}\n\n        for route in routes:\n            # extract inputs\n            route_input_names = self._extract_variables(env, [route[\"output\"], route[\"condition\"]])\n            input_types.update(route_input_names)\n\n            # extract outputs\n            output_types.update({route[\"output_name\"]: route[\"output_type\"]})\n\n        component.set_input_types(self, **{var: Any for var in input_types})\n        component.set_output_types(self, **output_types)\n\n    def to_dict(self) -> Dict[str, Any]:\n        \"\"\"\n        Serializes the component to a dictionary.\n\n        :returns:\n            Dictionary with serialized data.\n        \"\"\"\n        for route in self.routes:\n            # output_type needs to be serialized to a string\n            route[\"output_type\"] = serialize_type(route[\"output_type\"])\n        se_filters = {name: serialize_callable(filter_func) for name, filter_func in self.custom_filters.items()}\n        return default_to_dict(self, routes=self.routes, custom_filters=se_filters)\n\n    @classmethod\n    def from_dict(cls, data: Dict[str, Any]) -> \"ConditionalRouter\":\n        \"\"\"\n        Deserializes the component from a dictionary.\n\n        :param data:\n            The dictionary to deserialize from.\n        :returns:\n            The deserialized component.\n        \"\"\"\n        init_params = data.get(\"init_parameters\", {})\n        routes = init_params.get(\"routes\")\n        for route in routes:\n            # output_type needs to be deserialized from a string to a type\n            route[\"output_type\"] = deserialize_type(route[\"output_type\"])\n        for name, filter_func in init_params.get(\"custom_filters\", {}).items():\n            init_params[\"custom_filters\"][name] = deserialize_callable(filter_func) if filter_func else None\n        return default_from_dict(cls, data)\n\n    def run(self, **kwargs):\n        \"\"\"\n        Executes the routing logic.\n\n        Executes the routing logic by evaluating the specified boolean condition expressions for each route in the\n        order they are listed. The method directs the flow of data to the output specified in the first route whose\n        `condition` is True.\n\n        :param kwargs: All variables used in the `condition` expressed in the routes. When the component is used in a\n            pipeline, these variables are passed from the previous component's output.\n\n        :returns: A dictionary where the key is the `output_name` of the selected route and the value is the `output`\n            of the selected route.\n\n        :raises NoRouteSelectedException: If no `condition' in the routes is `True`.\n        :raises RouteConditionException: If there is an error parsing or evaluating the `condition` expression in the\n            routes.\n        \"\"\"\n        # Create a Jinja native environment to evaluate the condition templates as Python expressions\n        env = NativeEnvironment()\n        env.filters.update(self.custom_filters)\n\n        for route in self.routes:\n            try:\n                t = env.from_string(route[\"condition\"])\n                if t.render(**kwargs):\n                    # We now evaluate the `output` expression to determine the route output\n                    t_output = env.from_string(route[\"output\"])\n                    output = t_output.render(**kwargs)\n                    # and return the output as a dictionary under the output_name key\n                    return {route[\"output_name\"]: output}\n            except Exception as e:\n                raise RouteConditionException(f\"Error evaluating condition for route '{route}': {e}\") from e\n\n        raise NoRouteSelectedException(f\"No route fired. Routes: {self.routes}\")\n\n    def _validate_routes(self, routes: List[Dict]):\n        \"\"\"\n        Validates a list of routes.\n\n        :param routes: A list of routes.\n        \"\"\"\n        env = NativeEnvironment()\n        for route in routes:\n            try:\n                keys = set(route.keys())\n            except AttributeError:\n                raise ValueError(f\"Route must be a dictionary, got: {route}\")\n\n            mandatory_fields = {\"condition\", \"output\", \"output_type\", \"output_name\"}\n            has_all_mandatory_fields = mandatory_fields.issubset(keys)\n            if not has_all_mandatory_fields:\n                raise ValueError(\n                    f\"Route must contain 'condition', 'output', 'output_type' and 'output_name' fields: {route}\"\n                )\n            for field in [\"condition\", \"output\"]:\n                if not self._validate_template(env, route[field]):\n                    raise ValueError(f\"Invalid template for field '{field}': {route[field]}\")\n\n    def _extract_variables(self, env: NativeEnvironment, templates: List[str]) -> Set[str]:\n        \"\"\"\n        Extracts all variables from a list of Jinja template strings.\n\n        :param env: A Jinja environment.\n        :param templates: A list of Jinja template strings.\n        :returns: A set of variable names.\n        \"\"\"\n        variables = set()\n        for template in templates:\n            ast = env.parse(template)\n            variables.update(meta.find_undeclared_variables(ast))\n        return variables\n\n    def _validate_template(self, env: Environment, template_text: str):\n        \"\"\"\n        Validates a template string by parsing it with Jinja.\n\n        :param env: A Jinja environment.\n        :param template_text: A Jinja template string.\n        :returns: `True` if the template is valid, `False` otherwise.\n        \"\"\"\n        try:\n            env.parse(template_text)\n            return True\n        except TemplateSyntaxError:\n            return False\n", "path": "haystack/components/routers/conditional_router.py"}]}