
Most common position in a pipeline	Flexible
Mandatory init variables	"routes": A list of dictionaries defining routs (See the Overview section below)
Mandatory run variables	“**kwargs”: Input variables to evaluate in order to choose a specific route. See Variables section for more details.
Output variables	A dictionary containing one or more output names and values of the chosen route
API reference	Routers
GitHub link	https://github.com/deepset-ai/haystack/blob/main/haystack/components/routers/conditional_router.py

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 or list of expressions defining one or more output values.
'output_type': The expected type or list of types corresponding to each output (for example, str, List[int]).
- Note that this doesn't enforce the type conversion of the output. Instead, the output field is rendered using Jinja2, which automatically infers types. If you need to ensure the result is a string (for example, "123" instead of 123), wrap the Jinja expression in single quotes like this: output: "'{{message.text}}'". This ensures the rendered output is treated as a string by Jinja2.
'output_name': The name or list of names under which the output values are published. This is used to connect the router to other components in the pipeline.

Variables

The ConditionalRouter lets you define which variables are optional in your routing conditions.

from haystack.components.routers import ConditionalRouter

routes = [
    {
        "condition": '{{ path == "rag" }}',
        "output": "{{ question }}",
        "output_name": "rag_route",
        "output_type": str
    },
    {
        "condition": "{{ True }}",  # fallback route
        "output": "{{ question }}",
        "output_name": "default_route",
        "output_type": str
    }
]

# 'path' is optional, 'question' is required
router = ConditionalRouter(
    routes=routes,
    optional_variables=["path"]
)

The component only waits for the required inputs before running. If you use an optional variable in a condition but don't provide it at runtime, it’s evaluated as None, which generally does not raise an error but can affect the condition’s outcome.

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 its length and returns both the text and its character count.

If the query is too short, the pipeline returns a warning message and the character count, then stops.

If the query is long enough, the pipeline returns the original query and its character count, sends the query to the PromptBuilder, and then to the Generator to produce the final answer.

from haystack import Pipeline
from haystack.components.routers import ConditionalRouter
from haystack.components.builders.chat_prompt_builder import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage

# Two routes, each returning two outputs: the text and its length
routes = [
    {
        "condition": "{{ query|length > 10 }}",
        "output": ["{{ query }}", "{{ query|length }}"],
        "output_name": ["ok_query", "length"],
        "output_type": [str, int],
    },
    {
        "condition": "{{ query|length <= 10 }}",
        "output": ["query too short: {{ query }}", "{{ query|length }}"],
        "output_name": ["too_short_query", "length"],
        "output_type": [str, int],
    },
]

router = ConditionalRouter(routes=routes)

pipe = Pipeline()
pipe.add_component("router", router)
pipe.add_component(
    "prompt_builder",
    ChatPromptBuilder(
        template=[ChatMessage.from_user("Answer the following query: {{ query }}")],
        required_variables={"query"},
    ),
)
pipe.add_component("generator", OpenAIChatGenerator())

pipe.connect("router.ok_query", "prompt_builder.query")
pipe.connect("prompt_builder.prompt", "generator.messages")


# Short query: length ≤ 10 ⇒ fallback route fires.
print(pipe.run(data={"router": {"query": "Berlin"}}))
# {'router': {'too_short_query': 'query too short: Berlin', 'length': 6}}

# Long query: length > 10 ⇒ first route fires.
print(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