ConditionalRouter
ConditionalRouter routes your data through different paths down the pipeline by evaluating the conditions that you specified.
| Most common position in a pipeline | Flexible |
| Mandatory init variables | routes: A list of dictionaries defining routes (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 |
| Package name | haystack-ai |
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.
- 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
'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.
Usage
Basic routing
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
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)
result = router.run(streams=[1, 2, 3], query="Haystack")
print(result)
# {"enough_streams": [1, 2, 3]}
Multiple outputs per route
Each route can emit more than one output at a time. Pass lists to output, output_name, and output_type — all three must have the same length.
from haystack.components.routers import ConditionalRouter
routes = [
{
"condition": "{{ query|length > 10 }}",
"output": ["{{ query }}", "{{ query|length }}"],
"output_name": ["long_query", "char_count"],
"output_type": [str, int],
},
{
"condition": "{{ query|length <= 10 }}",
"output": ["{{ query }}", "{{ query|length }}"],
"output_name": ["short_query", "char_count"],
"output_type": [str, int],
},
]
router = ConditionalRouter(routes=routes)
result = router.run(query="Hello")
print(result)
# {'short_query': 'Hello', 'char_count': 5}
All outputs from the selected route are emitted together, so downstream components can consume any combination of them.
Variables
By default, every Jinja2 variable referenced in your route condition and output templates is required — the component won't run until all of them are provided. You can mark specific variables as optional using the optional_variables init parameter.
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"])
# 'path' provided — first route matches
print(router.run(question="What is RAG?", path="rag"))
# {'rag_route': 'What is RAG?'}
# 'path' omitted — evaluates as None, fallback route fires
print(router.run(question="What is RAG?"))
# {'default_route': 'What is RAG?'}
If an optional variable is not provided at runtime, it's evaluated as None, which generally does not raise an error but can affect the condition's outcome.
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?"}}))
# {
# 'router': {'length': 29},
# 'generator': {'replies': [ChatMessage(content='The capital of Italy is Rome (Italian: Roma).', role=<ChatRole.ASSISTANT: 'assistant'>)]}
# }
Configuration
Unsafe mode
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.
Custom filters
You can pass custom Jinja2 filter functions to use inside your route condition and output templates via the custom_filters init parameter.
from haystack.components.routers import ConditionalRouter
def first_word(value: str) -> str:
return value.split()[0] if value else ""
routes = [
{
"condition": '{{ query|first_word == "summarize" }}',
"output": "{{ query }}",
"output_name": "summarize_route",
"output_type": str,
},
{
"condition": "{{ True }}",
"output": "{{ query }}",
"output_name": "default_route",
"output_type": str,
},
]
router = ConditionalRouter(routes=routes, custom_filters={"first_word": first_word})
print(router.run(query="summarize this document"))
# {'summarize_route': 'summarize this document'}
print(router.run(query="what is the capital of France?"))
# {'default_route': 'what is the capital of France?'}
Output type validation
By default, ConditionalRouter does not verify that a route's output matches the declared output_type.
Set validate_output_type=True to enable this check which is useful to catch cases where a template didn't produce the type you expected.
from haystack.components.routers import ConditionalRouter
routes = [
{
"condition": "{{ True }}",
"output": "{{ value }}",
"output_name": "result",
"output_type": int,
},
]
# Without validation: a string passes through silently
router = ConditionalRouter(routes=routes)
print(router.run(value="not_a_number"))
# {'result': 'not_a_number'} — wrong type, no error raised
# With validation: type mismatch raises a ValueError
strict_router = ConditionalRouter(routes=routes, validate_output_type=True)
strict_router.run(value="not_a_number")
# ValueError: Route 'result' type doesn't match expected type
Additional References
📓 Tutorial: Building Fallbacks to Websearch with Conditional Routing