
Name	ConditionalRouter
Folder path	/routers/
Most common position in a pipeline	Flexible
Mandatory input variables	“**kwargs”: Input variables to evaluate in order to choose a specific route
Output variables	A dictionary containing the output name and value of the chosen route

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.

📘
Output Types
For security reasons, the ConditionalRouter can only return the following Python literal structures: strings, bytes, numbers, tuples, lists, dicts, sets, booleans, None and Ellipsis (...), as well as any combination of these structures. This means that Haystack classes such as ChatMessage, Document, or Answer can't be used as output types.

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.'], ...}

Overview

📘Output Types

Usage

On its own

In a pipeline

📘
Output Types