
pipeline 中的最常见位置	灵活
必需的初始化变量	"routes": 定义路由的字典列表（请参阅下面的概述部分）
强制运行变量	“**kwargs”: 用于评估以选择特定路由的输入变量。有关更多详细信息，请参阅变量部分。
输出变量	一个包含一个或多个输出名称和所选路由值的字典
API 参考	Routers (路由器)
GitHub 链接	https://github.com/deepset-ai/haystack/blob/main/haystack/components/routers/conditional_router.py

概述

要使用ConditionalRouter 您需要定义一个路由列表。
每个路由都是一个包含以下元素的字典

'condition': 一个 Jinja2 字符串表达式，用于确定是否选择该路由。
'output': 一个 Jinja2 表达式或表达式列表，用于定义一个或多个输出值。
'output_type': 对应于每个输出的预期类型或类型列表（例如，str, List[int]).
- 请注意，这不会强制执行输出的类型转换。相反，输出字段是使用 Jinja2 渲染的，Jinja2 会自动推断类型。如果您需要确保结果为字符串（例如，“123”而不是123），请将 Jinja 表达式包装在单引号中，如下所示output: "'{{message.text}}'"。这可以确保渲染后的输出被 Jinja2 视为字符串。
'output_name': 用于发布输出值的名称或名称列表。这用于将路由器连接到管道中的其他组件。

变量

该ConditionalRouter 允许您定义路由条件中哪些变量是可选的。

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"]
)

该组件仅在运行时等待必需的输入。如果您在条件中使用可选变量，但在运行时未提供它，则它将被评估为None，这通常不会引发错误，但可能会影响条件的结果。

不安全行为

该ConditionalRouter 默认情况下，它会使用 Jinja 安全地渲染所有规则的模板。这会限制输出类型为字符串、字节、数字、元组、列表、字典、集合、布尔值、None 和省略号 (...），以及这些结构的任何组合。

如果您想使用更多类型，例如ChatMessage, Document 或Answer，您必须通过设置unsafe 初始化参数来启用不安全模板的渲染。True.

请注意，这并不安全，并且可能导致远程代码执行，如果规则condition 或output 模板可以由最终用户自定义。

用法

单独使用

此组件主要用于管道中。

在此示例中，我们配置了两个路由。第一个路由将'streams' 值发送到'enough_streams'，前提是流的数量超过两个。反之，第二个路由将'streams' 发送到'insufficient_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]}

在 pipeline 中

下面是一个简单的管道示例，该管道根据查询长度进行路由，并返回文本及其字符数。

如果查询太短，管道将返回一条警告消息和字符数，然后停止。

如果查询足够长，管道将返回原始查询及其字符数，将查询发送到PromptBuilder，然后发送到 Generator 以生成最终答案。

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

其他参考资料

📓 教程：使用条件路由构建回退到 Web 搜索