ConditionalRouter

Routes data based on specific conditions.

Basic Information

• Type: haystack_integrations.routers.conditional_router.ConditionalRouter

Inputs

Parameter	Type	Default	Description
kwargs	Any		All variables used in the condition expressed in the routes. When the component is used in a pipeline, these variables are passed from the previous component's output.

Outputs

Parameter	Type	Default	Description

Overview

Work in Progress

Bear with us while we're working on adding pipeline examples and most common components connections.

Routes data based on specific conditions.

You define these conditions in a list of dictionaries called routes. Each dictionary in this list represents a single route. Each route has these four 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 type of the output data (for example, str, List[int]).
• output_name: The name you want to use to publish output. This name is used to connect the router to other components in the pipeline.In this example, we configure two routes. The first route sends the 'streams' value to 'enough_streams' if the stream count exceeds two. The second route directs 'streams' to 'insufficient_streams' if there are two or fewer streams.

In the pipeline setup, the Router connects to other components using the output names. For example, 'enough_streams' might connect to a component that processes streams, while 'insufficient_streams' might connect to a component that fetches more streams.

Here is a pipeline that uses ConditionalRouter and routes the fetched ByteStreams to different components depending on the number of streams fetched:

from typing import List
from haystack import Pipeline
from haystack.dataclasses import ByteStream
from haystack.components.routers import ConditionalRouter

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

pipe = Pipeline()
pipe.add_component("router", router)
...
pipe.connect("router.enough_streams", "some_component_a.streams")
pipe.connect("router.insufficient_streams", "some_component_b.streams_or_some_other_input")
...

Usage Example

components:
  ConditionalRouter:
    type: components.routers.conditional_router.ConditionalRouter
    init_parameters:

Parameters

Init Parameters

These are the parameters you can configure in Pipeline Builder:

Parameter	Type	Default	Description
routes	List[Route]		A list of dictionaries, each defining a route. Each route has these four 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 type of the output data (for example, str, List[int]). - output_name: The name you want to use to publish output. This name is used to connect the router to other components in the pipeline.
custom_filters	Optional[Dict[str, Callable]]	None	A dictionary of custom Jinja2 filters 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: "condition": "{{ my_var|my_filter == 'foo' }}".
unsafe	bool	False	Enable execution of arbitrary code in the Jinja template. This should only be used if you trust the source of the template as it can be lead to remote code execution.
validate_output_type	bool	False	Enable validation of routes' output. If a route output doesn't match the declared type a ValueError is raised running.
optional_variables	Optional[List[str]]	None	A list of variable names that are optional in your route conditions and outputs. If these variables are not provided at runtime, they will be set to None. This allows you to write routes that can handle missing inputs gracefully without raising errors. Example usage with a default fallback route in a Pipeline: python from haystack import Pipeline 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 } ] router = ConditionalRouter(routes, optional_variables=["path"]) pipe = Pipeline() pipe.add_component("router", router) # When 'path' is provided in the pipeline: result = pipe.run(data={"router": {"question": "What?", "path": "rag"}}) assert result["router"] == {"rag_route": "What?"} # When 'path' is not provided, fallback route is taken: result = pipe.run(data={"router": {"question": "What?"}}) assert result["router"] == {"default_route": "What?"} This pattern is particularly useful when: - You want to provide default/fallback behavior when certain inputs are missing - Some variables are only needed for specific routing conditions - You're building flexible pipelines where not all inputs are guaranteed to be present

Run Method Parameters

These are the parameters you can configure for the component's run() method. This means you can pass these parameters at query time through the API, in Playground, or when running a job. For details, see Modify Pipeline Parameters at Query Time.

Parameter	Type	Default	Description
kwargs	Any		All variables used in the condition expressed in the routes. When the component is used in a pipeline, these variables are passed from the previous component's output.