조건부 라우터

조건부 라우터 (ConditionalRouter)

ConditionalRouter는 여러분이 지정한 조건을 평가해서 파이프라인 아래쪽으로 데이터를 서로 다른 경로로 흘려 보내는 라우터예요. 쿼리 길이 같은 조건에 따라 검색을 거치게 할지, 아니면 바로 기본 응답을 내보낼지를 분기할 때 유용해요. 파이프라인에서 어디에 놓느냐가 정해져 있지 않은 유연한 컴포넌트예요.

출처: 공식문서

주요 정보

개요

ConditionalRouter를 사용하려면 경로(routes) 리스트를 정의해야 해요. 각 경로는 다음 요소를 가진 딕셔너리예요.

• 'condition': 경로가 선택될지 결정하는 Jinja2 문자열 표현식.
• 'output': 하나 이상의 출력 값을 정의하는 Jinja2 표현식 또는 표현식 리스트.
• 'output_type': 각 출력에 해당하는 기대 타입(예: str, list[int]).
  • 참고로 이 값은 출력의 타입 변환을 강제하지 않아요. 대신 output 필드는 타입을 자동 추론하는 Jinja2로 렌더돼요. 결과가 문자열인지 확인하고 싶다면(예: 123이 아니라 "123"), Jinja 표현식을 작은따옴표로 감싸면 돼요: output: "'{{message.text}}'". 이러면 렌더된 출력이 Jinja2에 의해 문자열로 취급돼요.
• 'output_name': 출력 값이 게시되는 이름 또는 이름 리스트. 라우터를 파이프라인의 다른 컴포넌트에 연결할 때 사용해요.

사용법

기본 라우팅

이 예시에서는 경로 두 개를 설정해요. 첫 번째 경로는 스트림 개수가 2를 넘으면 'streams' 값을 'enough_streams'로 보내요. 반대로 두 번째 경로는 스트림이 2 이하일 때 'streams'를 'insufficient_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]}

경로당 여러 출력

각 경로는 한 번에 하나 이상의 출력을 내보낼 수 있어요. output, output_name, output_type에 리스트를 넘기면 되는데, 세 리스트 모두 길이가 같아야 해요.

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}

선택된 경로의 모든 출력은 함께 내보내져서, 다운스트림 컴포넌트가 출력의 어떤 조합이든 사용할 수 있어요.

변수 (Variables)

기본적으로 경로의 condition과 output 템플릿에 참조된 모든 Jinja2 변수는 필수예요 — 모두 제공될 때까지 컴포넌트가 실행되지 않아요. optional_variables 초기화 파라미터로 특정 변수를 선택 사항으로 표시할 수 있어요.

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?'}

선택 변수가 런타임에 제공되지 않으면 None으로 평가돼요. 일반적으로 오류를 일으키지는 않지만 조건의 결과에 영향을 줄 수 있어요.

파이프라인 안에서 쓰기

아래는 쿼리 길이에 따라 경로를 나누고, 텍스트와 문자 개수를 함께 돌려주는 간단한 파이프라인 예시예요.

쿼리가 너무 짧으면 파이프라인은 경고 메시지와 문자 개수를 반환하고 멈춰요.

쿼리가 충분히 길면 파이프라인은 원래 쿼리와 문자 개수를 반환하고, 쿼리를 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?"}}))
# {
#   'router': {'length': 29},
#   'generator': {'replies': [ChatMessage(content='The capital of Italy is Rome (Italian: Roma).', role=<ChatRole.ASSISTANT: 'assistant'>)]}
# }

설정 (Configuration)

안전하지 않은 모드 (Unsafe mode)

ConditionalRouter는 내부적으로 모든 규칙 템플릿을 Jinja로 렌더링하는데, 기본적으로 이는 안전한 동작이에요. 다만 출력 타입이 문자열, 바이트, 숫자, 튜플, 리스트, 딕셔너리, 집합, 불리언, None, Ellipsis(...), 그리고 이 구조들의 조합으로 제한돼요.

ChatMessage, Document, Answer 같은 더 많은 타입을 쓰려면 unsafe 초기화 인자를 True로 설정해 안전하지 않은 템플릿 렌더링을 켜야 해요.

이것은 안전하지 않은 동작이며, 규칙의 condition이나 output 템플릿이 최종 사용자에 의해 커스터마이징될 수 있다면 원격 코드 실행으로 이어질 수 있으니 주의하세요.

커스텀 필터

custom_filters 초기화 파라미터를 통해 경로의 condition과 output 템플릿 안에서 사용할 커스텀 Jinja2 필터 함수를 넘길 수 있어요.

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?'}

출력 타입 검증

기본적으로 ConditionalRouter는 경로의 출력이 선언된 output_type과 일치하는지 확인하지 않아요. validate_output_type=True로 설정하면 이 검사를 켤 수 있는데, 템플릿이 기대한 타입을 만들지 못한 경우를 잡는 데 유용해요.

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

더 알아보기

• 튜토리얼: Building Fallbacks to Websearch with Conditional Routing
• Haystack 라우터 API 레퍼런스