Overview

To use ConditionalRouter you need to define a list of routes.
Each route is a dictionary with the following elements:

• 'condition': A Jinja2 string expression that determines if the route is selected.
• 'output': A Jinja2 expression defining the route's output value.
• 'output_type': The expected type of the output data (for example, str or List[int]). This doesn't change the actual output type and is only needed to validate the connection with other components.
• '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.

Unsafe behaviour

The ConditionalRouter internally renders all the rules' templates using Jinja, by default this is a safe behaviour. Though it limits the output types to strings, bytes, numbers, tuples, lists, dicts, sets, booleans, None and Ellipsis (...), as well as any combination of these structures.

If you want to use more types like ChatMessage, Document or Answer you must enable rendering of unsafe templates by setting the unsafe init argument to True.

Beware that this is unsafe and can lead to remote code execution if a rule condition or output templates are customizable by the end user.

Usage

On its own

This component is primarily meant to be used in pipelines.

In this example, we configure two routes. The first route sends the 'streams' value to 'enough_streams' if the stream count exceeds two. Conversely, the second route directs 'streams' to 'insufficient_streams' when there are two or fewer streams.

from haystack.components.routers import ConditionalRouter
from typing import List

routes = [
    {
        "condition": "{{streams|length > 2}}",
        "output": "{{streams}}",
        "output_name": "enough_streams",
        "output_type": List[int],
    },
    {
        "condition": "{{streams|length <= 2}}",
        "output": "{{streams}}",
        "output_name": "insufficient_streams",
        "output_type": List[int],
    },
]

router = ConditionalRouter(routes)

kwargs = {"streams": [1, 2, 3], "query": "Haystack"}
result = router.run(**kwargs)

print(result)
# {"enough_streams": [1, 2, 3]}

In a pipeline

Below is an example of a simple pipeline that routes a query based on length.

If the query is too short, you will see a warning message, and the process will end.
If the query is long enough, it is first forwarded to the PromptBuilder, then to the Generator, which generates the final answer.

from haystack import Pipeline
from haystack.components.routers import ConditionalRouter
from haystack.components.builders.prompt_builder import PromptBuilder
from haystack.components.generators import OpenAIGenerator

routes = [
        {
            "condition": "{{query|length > 10}}",
            "output": "{{query}}",
            "output_name": "ok_query",
            "output_type": str,
        },
        {
            "condition": "{{query|length <= 10}}",
            "output": "query is too short: {{query}}",
            "output_name": "too_short_query",
            "output_type": str,
        },
    ]
router = ConditionalRouter(routes=routes)

pipe = Pipeline()
pipe.add_component("router", router)
pipe.add_component("prompt_builder", PromptBuilder("Answer the following query. {{query}}"))
pipe.add_component("generator", OpenAIGenerator())
pipe.connect("router.ok_query", "prompt_builder.query")
pipe.connect("prompt_builder", "generator")

pipe.run(data={"router": {"query": "Berlin"}})
# {'router': {'too_short_query': 'query is too short: Berlin'}}

pipe.run(data={"router": {"query": "What is the capital of Italy?"}})
# {'generator': {'replies': ['The capital of Italy is Rome.'], ...}

Additional References

📓 Tutorial: Building Fallbacks to Websearch with Conditional Routing